Merge pull request #64 from sjtug/wip-implement-concurrent

implement concurrent_limit

Gopkg.toml

@@ -52,3 +52,6 @@
 [prune]
   go-tests = true
   unused-packages = true
+
+[[constraint]]
+  name = "github.com/davecgh/go-spew"

config.example.yaml

@@ -1,5 +1,6 @@
 interval: 3 # Interval between pollings
 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
 
@@ -16,12 +17,15 @@ repos:
     - type: shell_script
       script: rsync -av rsync://rsync.chiark.greenend.org.uk/ftp/users/sgtatham/putty-website-mirror/ /tmp/putty
       name: putty
+      interval: 600
     - type: shell_script
       script: bash -c 'printenv | grep ^LUG'
       name: printenv
       any_option: any_value
       any_switch: true # This will be set to 1
       any_switch_2: false # unset
+      interval: 10
     - type: external
       name: ubuntu
       proxy_to: http://ftp.sjtu.edu.cn/ubuntu/
+      # Since interval is not set for this target, this will only be triggered at startup

pkg/config/config.go

@@ -31,6 +31,8 @@ type Config struct {
 	Interval int
 	// LogLevel: 0-5 is acceptable
 	LogLevel log.Level
+	// ConcurrentLimit: how many worker can run at the same time
+	ConcurrentLimit int `mapstructure:"concurrent_limit"`
 	// LogStashConfig represents configurations for logstash
 	LogStashConfig LogStashConfig `mapstructure:"logstash"`
 	// ExporterAddr is the address to expose metrics, :8080 for default
@@ -49,6 +51,7 @@ func init() {
 	CfgViper.SetDefault("loglevel", 4)
 	CfgViper.SetDefault("json_api.address", ":7001")
 	CfgViper.SetDefault("exporter_address", ":8080")
+	CfgViper.SetDefault("concurrent_limit", 5)
 }
 
 // Parse creates config from a reader
@@ -66,6 +69,9 @@ func (c *Config) Parse(in io.Reader) (err error) {
 		if c.LogLevel < 0 || c.LogLevel > 5 {
 			return errors.New("loglevel must be 0-5")
 		}
+		if c.ConcurrentLimit <= 0 {
+			return errors.New("concurrent limit must be positive")
+		}
 	}
 	for _, repo := range c.Repos {
 		for _, v := range repo {

pkg/config/config_test.go

@@ -10,6 +10,7 @@ import (
 func TestParseConfig(t *testing.T) {
 	const testStr = `interval: 25
 loglevel: 5 # 1 - 5
+concurrent_limit: 6
 repos:
 - type: shell_script
   script: rsync -av rsync://rsync.chiark.greenend.org.uk/ftp/users/sgtatham/putty-website-mirror/ /tmp/putty
@@ -24,6 +25,7 @@ repos:
 	asrt.Equal(25, c.Interval)
 	asrt.Equal(5, int(c.LogLevel))
 	asrt.Equal(1, len(c.Repos))
+	asrt.Equal(6, c.ConcurrentLimit)
 	asrt.EqualValues("shell_script", c.Repos[0]["type"])
 	asrt.EqualValues("rsync -av rsync://rsync.chiark.greenend.org.uk/ftp/users/sgtatham/putty-website-mirror/ /tmp/putty", c.Repos[0]["script"])
 	asrt.EqualValues("/mnt/putty", c.Repos[0]["path"])
@@ -81,4 +83,18 @@ repos:
 	err = c.Parse(strings.NewReader(testStr))
 
 	asrt.Equal("loglevel must be 0-5", err.Error())
+
+	testStr = `interval: 25
+loglevel: 4
+concurrent_limit: 0
+repos:
+- type: shell_script
+  script: rsync -av rsync://rsync.chiark.greenend.org.uk/ftp/users/sgtatham/putty-website-mirror/ /tmp/putty
+  name: putty
+  path: /mnt/putty
+`
+	c = Config{}
+	err = c.Parse(strings.NewReader(testStr))
+
+	asrt.Equal("concurrent limit must be positive", err.Error())
 }

pkg/manager/manager.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/sirupsen/logrus"
 
+	"github.com/davecgh/go-spew/spew"
 	"github.com/sjtug/lug/pkg/config"
 	"github.com/sjtug/lug/pkg/worker"
 )
@@ -33,7 +34,9 @@ type Manager struct {
 	controlChan           chan int
 	finishChan            chan int
 	running               bool
-	logger                *logrus.Entry
+	// storing index of worker to launch
+	pendingQueue []int
+	logger       *logrus.Entry
 }
 
 // Status holds the status of a manager and its workers
@@ -65,6 +68,46 @@ func NewManager(config *config.Config) (*Manager, error) {
 	return &newManager, nil
 }
 
+func (m *Manager) isAlreadyInPendingQueue(workerIdx int) bool {
+	for _, wk := range m.pendingQueue {
+		if wk == workerIdx {
+			return true
+		}
+	}
+	return false
+}
+
+func (m *Manager) launchWorkerFromPendingQueue(max_allowed int) {
+	if max_allowed <= 0 {
+		return
+	}
+	var new_idx int
+	if max_allowed > len(m.pendingQueue) {
+		new_idx = len(m.pendingQueue)
+	} else {
+		new_idx = max_allowed
+	}
+	m.logger.WithFields(logrus.Fields{
+		"event":         "launch_worker_from_pending_queue",
+		"max_allowed":   max_allowed,
+		"new_idx":       new_idx,
+		"pending_queue": spew.Sprint(m.pendingQueue),
+	}).Debug("launch worker from pending queue")
+	to_launch := m.pendingQueue[:new_idx]
+	m.pendingQueue = m.pendingQueue[new_idx:]
+
+	for _, w_idx := range to_launch {
+		w := m.workers[w_idx]
+		wConfig := w.GetConfig()
+		m.logger.WithFields(logrus.Fields{
+			"event":              "trigger_sync",
+			"target_worker_name": wConfig["name"],
+		}).Infof("trigger sync for worker %s from pendingQueue", wConfig["name"])
+		m.workersLastInvokeTime[w_idx] = time.Now()
+		w.TriggerSync()
+	}
+}
+
 // Run will block current routine
 func (m *Manager) Run() {
 	m.logger.Debugf("%p", m)
@@ -82,6 +125,7 @@ func (m *Manager) Run() {
 		case <-c:
 			if m.running {
 				m.logger.WithField("event", "poll_start").Info("Start polling workers")
+				running_worker_cnt := 0
 				for i, w := range m.workers {
 					wStatus := w.GetStatus()
 					m.logger.WithFields(logrus.Fields{
@@ -92,21 +136,25 @@ func (m *Manager) Run() {
 						"target_worker_last_finished": wStatus.LastFinished,
 					})
 					if !wStatus.Idle {
+						running_worker_cnt++
 						continue
 					}
 					wConfig := w.GetConfig()
 					elapsed := time.Since(m.workersLastInvokeTime[i])
-					sec2sync, _ := wConfig["interval"].(int)
-					if elapsed > time.Duration(sec2sync)*time.Second {
+					sec2sync, ok := wConfig["interval"].(int)
+					if !ok {
+						sec2sync = 31536000 // if "interval" is not specified, then worker will launch once a year
+					}
+					if !m.isAlreadyInPendingQueue(i) && elapsed > time.Duration(sec2sync)*time.Second {
 						m.logger.WithFields(logrus.Fields{
-							"event":                  "trigger_sync",
+							"event":                  "trigger_pending",
 							"target_worker_name":     wConfig["name"],
 							"target_worker_interval": sec2sync,
-						}).Infof("Interval of w %s (%d sec) elapsed, trigger it to sync", wConfig["name"], sec2sync)
-						m.workersLastInvokeTime[i] = time.Now()
-						w.TriggerSync()
+						}).Infof("Interval of w %s (%d sec) elapsed, send it to pendingQueue", wConfig["name"], sec2sync)
+						m.pendingQueue = append(m.pendingQueue, i)
 					}
 				}
+				m.launchWorkerFromPendingQueue(m.config.ConcurrentLimit - running_worker_cnt)
 				m.logger.WithField("event", "poll_end").Info("Stop polling workers")
 			}
 		case sig, ok := <-m.controlChan:

pkg/worker/external_worker.go

@@ -34,7 +34,7 @@ func (ew *ExternalWorker) GetStatus() Status {
 	return Status{
 		Result:       true,
 		LastFinished: time.Now(),
-		Idle:         false,
+		Idle:         true,
 		Stdout:       []string{},
 		Stderr:       []string{},
 	}

pkg/worker/worker_test.go

@@ -31,7 +31,7 @@ func TestNewExternalWorker(t *testing.T) {
 
 	status := w.GetStatus()
 	asrt.True(status.Result)
-	asrt.False(status.Idle)
+	asrt.True(status.Idle)
 	asrt.NotNil(status.Stderr)
 	asrt.NotNil(status.Stdout)
 }