Remember last finished time and result

pkg/manager/manager.go

@@ -4,13 +4,13 @@ package manager
 import (
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"github.com/davecgh/go-spew/spew"
+	"io"
 	"os"
 	"time"
 
 	"github.com/sirupsen/logrus"
 
-	"github.com/davecgh/go-spew/spew"
 	"github.com/sjtug/lug/pkg/config"
 	"github.com/sjtug/lug/pkg/worker"
 )
@@ -51,7 +51,9 @@ type Status struct {
 }
 
 type WorkerCheckPoint struct {
-	LastInvokeTime time.Time `json:"last_invoke_time"`
+	LastInvokeTime time.Time  `json:"last_invoke_time"`
+	LastFinished   *time.Time `json:"last_finished,omitempty"`
+	Result         bool       `json:"result,omitempty"`
 }
 
 type CheckPoint struct {
@@ -64,9 +66,9 @@ func fromCheckpoint(checkpointFile string) (*CheckPoint, error) {
 	if err != nil {
 		return nil, err
 	}
-	defer jsonFile.Close()
+	defer func() { _ = jsonFile.Close() }()
 
-	data, err := ioutil.ReadAll(jsonFile)
+	data, err := io.ReadAll(jsonFile)
 	if err != nil {
 		return nil, err
 	}
@@ -81,6 +83,21 @@ func fromCheckpoint(checkpointFile string) (*CheckPoint, error) {
 	return &checkpoint, nil
 }
 
+func workerFromCheckpoint(repoConfig config.RepoConfig, checkpoint *CheckPoint, name string, lastInvokeTime time.Time) (worker.Worker, error) {
+	if checkpoint == nil {
+		return worker.NewWorker(repoConfig, lastInvokeTime, true)
+	}
+	info, ok := checkpoint.WorkerInfo[name]
+	if !ok {
+		return worker.NewWorker(repoConfig, lastInvokeTime, true)
+	}
+
+	if info.LastFinished == nil {
+		return worker.NewWorker(repoConfig, lastInvokeTime, info.Result)
+	}
+	return worker.NewWorker(repoConfig, *info.LastFinished, info.Result)
+}
+
 // NewManager creates a new manager with attached workers from config
 func NewManager(config *config.Config) (*Manager, error) {
 	logger := logrus.WithField("manager", "")
@@ -89,8 +106,8 @@ func NewManager(config *config.Config) (*Manager, error) {
 	if err != nil {
 		logger.Info("failed to parse checkpoint file")
 	} else {
-		for worker, info := range checkpoint.WorkerInfo {
-			workersLastInvokeTime[worker] = info.LastInvokeTime
+		for name, info := range checkpoint.WorkerInfo {
+			workersLastInvokeTime[name] = info.LastInvokeTime
 		}
 	}
 	newManager := Manager{
@@ -110,7 +127,7 @@ func NewManager(config *config.Config) (*Manager, error) {
 		if _, ok := newManager.workersLastInvokeTime[name]; !ok {
 			newManager.workersLastInvokeTime[name] = time.Now().AddDate(-1, 0, 0)
 		}
-		w, err := worker.NewWorker(repoConfig, newManager.workersLastInvokeTime[name])
+		w, err := workerFromCheckpoint(repoConfig, checkpoint, name, newManager.workersLastInvokeTime[name])
 		if err != nil {
 			return nil, err
 		}
@@ -121,17 +138,27 @@ func NewManager(config *config.Config) (*Manager, error) {
 
 func (m *Manager) checkpoint() error {
 	ckptObj := &CheckPoint{WorkerInfo: make(map[string]WorkerCheckPoint)}
-	for k, t := range m.workersLastInvokeTime {
-		ckptObj.WorkerInfo[k] = WorkerCheckPoint{
-			LastInvokeTime: t,
+	for _, w := range m.workers {
+		name := w.GetConfig()["name"].(string)
+		status := w.GetStatus()
+		lastInvokeTime, ok := m.workersLastInvokeTime[name]
+		if !ok {
+			lastInvokeTime = time.Now().AddDate(-1, 0, 0)
+		}
+
+		ckptObj.WorkerInfo[name] = WorkerCheckPoint{
+			LastInvokeTime: lastInvokeTime,
+			Result:         status.Result,
+			LastFinished:   &status.LastFinished,
 		}
 	}
+
 	file, err := json.MarshalIndent(ckptObj, "", "  ")
 	if err != nil {
 		return err
 	}
 	ckpt := fmt.Sprintf("%s.tmp", m.config.Checkpoint)
-	err = ioutil.WriteFile(ckpt, file, 0644)
+	err = os.WriteFile(ckpt, file, 0644)
 	if err != nil {
 		return err
 	}
@@ -193,7 +220,13 @@ func (m *Manager) Run() {
 		}).Debugf("Calling RunSync() to w %s", w.GetConfig()["name"])
 		go w.RunSync()
 	}
-	m.checkpoint()
+	err := m.checkpoint()
+	if err != nil {
+		m.logger.WithFields(logrus.Fields{
+			"event": "checkpoint_failed",
+			"error": err,
+		}).Error("Failed to checkpoint")
+	}
 	for {
 		// wait until config.Interval seconds has elapsed
 		select {
@@ -237,7 +270,13 @@ func (m *Manager) Run() {
 				// Here we do not checkpoint very concisely (e.g. every time after a successful sync).
 				// We just want to minimize re-sync after restarting lug.
 				if shouldCheckpoint {
-					m.checkpoint()
+					err := m.checkpoint()
+					if err != nil {
+						m.logger.WithFields(logrus.Fields{
+							"event": "checkpoint_failed",
+							"error": err,
+						}).Error("Failed to checkpoint")
+					}
 				}
 			}
 		case sig, ok := <-m.controlChan:

pkg/worker/worker.go

@@ -34,7 +34,7 @@ type Status struct {
 }
 
 // NewWorker generates a worker by config and log.
-func NewWorker(cfg config.RepoConfig, lastFinished time.Time) (Worker, error) {
+func NewWorker(cfg config.RepoConfig, lastFinished time.Time, Result bool) (Worker, error) {
 	if syncType, ok := cfg["type"]; ok {
 		switch syncType {
 		case "rsync":
@@ -44,7 +44,7 @@ func NewWorker(cfg config.RepoConfig, lastFinished time.Time) (Worker, error) {
 			w, err := NewExecutorInvokeWorker(
 				newShellScriptExecutor(cfg),
 				Status{
-					Result:       true,
+					Result:       Result,
 					LastFinished: lastFinished,
 					Idle:         true,
 					Stdout:       make([]string, 0),

pkg/worker/worker_test.go

@@ -24,12 +24,12 @@ func TestNewExternalWorker(t *testing.T) {
 		"blahblah": "foobar",
 		"type":     "external",
 	}
-	_, err := NewWorker(c, time.Now())
+	_, err := NewWorker(c, time.Now(), true)
 	// worker with no name is not allowed
 	asrt.NotNil(err)
 
 	c["name"] = "test_external"
-	w, err := NewWorker(c, time.Now())
+	w, err := NewWorker(c, time.Now(), true)
 	// config with name and dummy kv pairs should be allowed
 	asrt.Nil(err)
 
@@ -48,7 +48,7 @@ func TestNewShellScriptWorker(t *testing.T) {
 
 	asrt := assert.New(t)
 
-	w, _ := NewWorker(c, time.Now())
+	w, _ := NewWorker(c, time.Now(), true)
 
 	asrt.Equal(true, w.GetStatus().Result)
 	asrt.Equal("shell_script", w.GetConfig()["type"])
@@ -187,7 +187,7 @@ func TestShellScriptWorkerArgParse(t *testing.T) {
 		"name":   "shell",
 		"script": "wc -l /proc/stat",
 	}
-	w, err := NewWorker(c, time.Now())
+	w, err := NewWorker(c, time.Now(), true)
 
 	asrt := assert.New(t)
 	asrt.Nil(err)