manager: support checkpoint (#75)

.gitignore

@@ -6,6 +6,7 @@
 # Compiled binary files and configuration
 /lug
 config.yaml
+checkpoint.json
 
 # Folders
 _obj

config.example.yaml

@@ -3,6 +3,7 @@ loglevel: 5 # 1-5
 concurrent_limit: 1 # Maximum worker that can run at the same time
 # Prometheus metrics are exposed at http://exporter_address/metrics
 exporter_address: :8081
+checkpoint: checkpoint.json
 
 #logstash:
 #   address: listener.logz.io:5050 # logstash sink. Lug will send all logs to this address

pkg/config/config.go

@@ -39,6 +39,8 @@ type Config struct {
 	ExporterAddr string `mapstructure:"exporter_address"`
 	// JsonAPIConfig specifies configuration of JSON restful API
 	JsonAPIConfig JsonAPIConfig `mapstructure:"json_api"`
+	// Worker sync checkpoint path
+	Checkpoint string `mapstructure:"checkpoint"`
 	// Config for each repo is represented as an array of RepoConfig. Nested structure is disallowed
 	Repos []RepoConfig
 	// A dummy section that will not be used in our program.

pkg/manager/manager.go

@@ -2,6 +2,10 @@
 package manager
 
 import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -30,7 +34,7 @@ const (
 type Manager struct {
 	config                *config.Config
 	workers               []worker.Worker
-	workersLastInvokeTime []time.Time
+	workersLastInvokeTime map[string]time.Time
 	controlChan           chan int
 	finishChan            chan int
 	running               bool
@@ -46,16 +50,57 @@ type Status struct {
 	WorkerStatus map[string]worker.Status
 }
 
+type WorkerCheckPoint struct {
+	LastInvokeTime time.Time
+}
+
+type CheckPoint struct {
+	workerInfo map[string]WorkerCheckPoint
+}
+
+// fromCheckpoint laods last invoke time from json
+func fromCheckpoint(checkpointFile string) (*CheckPoint, error) {
+	jsonFile, err := os.Open(checkpointFile)
+	if err != nil {
+		return nil, err
+	}
+	defer jsonFile.Close()
+
+	data, err := ioutil.ReadAll(jsonFile)
+	if err != nil {
+		return nil, err
+	}
+
+	var checkpoint CheckPoint
+
+	err = json.Unmarshal(data, &checkpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	return &checkpoint, nil
+}
+
 // NewManager creates a new manager with attached workers from config
 func NewManager(config *config.Config) (*Manager, error) {
+	logger := logrus.WithField("manager", "")
+	checkpoint, err := fromCheckpoint(config.Checkpoint)
+	workersLastInvokeTime := make(map[string]time.Time)
+	if err != nil {
+		logger.Info("failed to parse checkpoint file")
+	} else {
+		for worker, info := range checkpoint.workerInfo {
+			workersLastInvokeTime[worker] = info.LastInvokeTime
+		}
+	}
 	newManager := Manager{
 		config:                config,
 		workers:               []worker.Worker{},
-		workersLastInvokeTime: []time.Time{},
+		workersLastInvokeTime: workersLastInvokeTime,
 		controlChan:           make(chan int),
 		finishChan:            make(chan int),
 		running:               true,
-		logger:                logrus.WithField("manager", ""),
+		logger:                logger,
 	}
 	for _, repoConfig := range config.Repos {
 		if disabled, ok := repoConfig["disabled"].(bool); ok && disabled {
@@ -66,11 +111,28 @@ func NewManager(config *config.Config) (*Manager, error) {
 			return nil, err
 		}
 		newManager.workers = append(newManager.workers, w)
-		newManager.workersLastInvokeTime = append(newManager.workersLastInvokeTime, time.Now().AddDate(-1, 0, 0))
+		name, _ := repoConfig["name"].(string)
+		if _, ok := newManager.workersLastInvokeTime[name]; !ok {
+			newManager.workersLastInvokeTime[name] = time.Now().AddDate(-1, 0, 0)
+		}
 	}
 	return &newManager, nil
 }
 
+func (m *Manager) checkpoint() error {
+	file, _ := json.MarshalIndent(m.workersLastInvokeTime, "", "  ")
+	ckpt := fmt.Sprintf("%s.tmp", m.config.Checkpoint)
+	err := ioutil.WriteFile(ckpt, file, 0644)
+	if err != nil {
+		return err
+	}
+	err = os.Rename(ckpt, m.config.Checkpoint)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func (m *Manager) isAlreadyInPendingQueue(workerIdx int) bool {
 	for _, wk := range m.pendingQueue {
 		if wk == workerIdx {
@@ -106,7 +168,7 @@ func (m *Manager) launchWorkerFromPendingQueue(max_allowed int) {
 			"event":              "trigger_sync",
 			"target_worker_name": wConfig["name"],
 		}).Infof("trigger sync for worker %s from pendingQueue", wConfig["name"])
-		m.workersLastInvokeTime[w_idx] = time.Now()
+		m.workersLastInvokeTime[wConfig["name"].(string)] = time.Now()
 		w.TriggerSync()
 	}
 }
@@ -129,6 +191,7 @@ func (m *Manager) Run() {
 			if m.running {
 				m.logger.WithField("event", "poll_start").Info("Start polling workers")
 				running_worker_cnt := 0
+				shouldCheckpoint := false
 				for i, w := range m.workers {
 					wStatus := w.GetStatus()
 					m.logger.WithFields(logrus.Fields{
@@ -143,7 +206,7 @@ func (m *Manager) Run() {
 						continue
 					}
 					wConfig := w.GetConfig()
-					elapsed := time.Since(m.workersLastInvokeTime[i])
+					elapsed := time.Since(m.workersLastInvokeTime[wConfig["name"].(string)])
 					sec2sync, ok := wConfig["interval"].(int)
 					if !ok {
 						sec2sync = 31536000 // if "interval" is not specified, then worker will launch once a year
@@ -155,10 +218,17 @@ func (m *Manager) Run() {
 							"target_worker_interval": sec2sync,
 						}).Infof("Interval of w %s (%d sec) elapsed, send it to pendingQueue", wConfig["name"], sec2sync)
 						m.pendingQueue = append(m.pendingQueue, i)
+						shouldCheckpoint = true
 					}
 				}
 				m.launchWorkerFromPendingQueue(m.config.ConcurrentLimit - running_worker_cnt)
 				m.logger.WithField("event", "poll_end").Info("Stop polling workers")
+
+				// 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()
+				}
 			}
 		case sig, ok := <-m.controlChan:
 			if ok {