Merge pull request #395 from Praqma/resourcePool

limit the number of parallel go routines, fixes #391

internal/app/decision_maker.go

@@ -25,16 +25,24 @@ func buildState(s *state) *currentState {
 	cs := newCurrentState()
 	rel := getHelmReleases(s)
 
-	var wg sync.WaitGroup
+	wg := sync.WaitGroup{}
+	sem := make(chan struct{}, resourcePool)
+
 	for _, r := range rel {
+		// aquire
+		sem <- struct{}{}
 		wg.Add(1)
-		go func(c *currentState, r helmRelease, wg *sync.WaitGroup) {
-			c.Lock()
-			defer c.Unlock()
-			defer wg.Done()
+		go func(r helmRelease) {
+			cs.Lock()
+			defer func() {
+				cs.Unlock()
+				wg.Done()
+				// release
+				<-sem
+			}()
 			r.HelmsmanContext = getReleaseContext(r.Name, r.Namespace)
-			c.releases[r.key()] = r
-		}(cs, r, &wg)
+			cs.releases[r.key()] = r
+		}(r)
 	}
 	wg.Wait()
 	return cs
@@ -45,10 +53,19 @@ func (cs *currentState) makePlan(s *state) *plan {
 	p := createPlan()
 
 	wg := sync.WaitGroup{}
+	sem := make(chan struct{}, resourcePool)
+
 	for _, r := range s.Apps {
 		r.checkChartDepUpdate()
+		sem <- struct{}{}
 		wg.Add(1)
-		go cs.decide(r, s, p, &wg)
+		go func(r *release) {
+			defer func() {
+				wg.Done()
+				<-sem
+			}()
+			cs.decide(r, s, p)
+		}(r)
 	}
 	wg.Wait()
 
@@ -57,8 +74,7 @@ func (cs *currentState) makePlan(s *state) *plan {
 
 // decide makes a decision about what commands (actions) need to be executed
 // to make a release section of the desired state come true.
-func (cs *currentState) decide(r *release, s *state, p *plan, wg *sync.WaitGroup) {
-	defer wg.Done()
+func (cs *currentState) decide(r *release, s *state, p *plan) {
 	// check for presence in defined targets or groups
 	if !r.isConsideredToRun(s) {
 		p.addDecision("Release [ "+r.Name+" ] ignored", r.Priority, ignored)
@@ -149,13 +165,20 @@ func (cs *currentState) getHelmsmanReleases(s *state) map[string]map[string]bool
 	)
 	const outputFmt = "custom-columns=NAME:.metadata.name,CTX:.metadata.labels.HELMSMAN_CONTEXT"
 	releases := make(map[string]map[string]bool)
+	sem := make(chan struct{}, resourcePool)
 	storageBackend := s.Settings.StorageBackend
 
 	for ns := range s.Namespaces {
+		// aquire
+		sem <- struct{}{}
 		wg.Add(1)
 		go func(ns string) {
 			var lines []string
-			defer wg.Done()
+			defer func() {
+				wg.Done()
+				// release
+				<-sem
+			}()
 
 			cmd := kubectl([]string{"get", storageBackend, "-n", ns, "-l", "MANAGED-BY=HELMSMAN", "-o", outputFmt, "--no-headers"}, "Getting Helmsman-managed releases")
 			result := cmd.exec()

internal/app/decision_maker_test.go

@@ -2,7 +2,6 @@ package app
 
 import (
 	"reflect"
-	"sync"
 	"testing"
 )
 
@@ -211,11 +210,8 @@ func Test_decide(t *testing.T) {
 				tt.args.s.TargetMap[target] = true
 			}
 			outcome := plan{}
-			wg := sync.WaitGroup{}
-			wg.Add(1)
 			// Act
-			cs.decide(tt.args.r, tt.args.s, &outcome, &wg)
-			wg.Wait()
+			cs.decide(tt.args.r, tt.args.s, &outcome)
 			got := outcome.Decisions[0].Type
 			t.Log(outcome.Decisions[0].Description)
 
@@ -297,10 +293,7 @@ func Test_decide_group(t *testing.T) {
 				tt.args.s.GroupMap[group] = true
 			}
 			outcome := plan{}
-			wg := sync.WaitGroup{}
-			wg.Add(1)
-			cs.decide(tt.args.r, tt.args.s, &outcome, &wg)
-			wg.Wait()
+			cs.decide(tt.args.r, tt.args.s, &outcome)
 			got := outcome.Decisions[0].Type
 			t.Log(outcome.Decisions[0].Description)
 

internal/app/helm_time.go

@@ -8,6 +8,7 @@ import (
 
 const ctLayout = "2006-01-02 15:04:05.000000000 -0700 MST"
 const ctLayout2 = "2006-01-02 15:04:05.000000000 -0700 -0700"
+
 var nilTime = (time.Time{}).UnixNano()
 
 type HelmTime struct {

internal/app/main.go

@@ -9,6 +9,7 @@ const (
 	appVersion         = "v3.0.1"
 	tempFilesDir       = ".helmsman-tmp"
 	defaultContextName = "default"
+	resourcePool       = 10
 )
 
 var (

internal/app/release.go

@@ -143,10 +143,18 @@ func (r *release) validate(appLabel string, names map[string]map[string]bool, s
 func validateReleaseCharts(s *state) error {
 	var fail bool
 	wg := sync.WaitGroup{}
+	sem := make(chan struct{}, resourcePool)
 	c := make(chan string, len(s.Apps))
 	for app, r := range s.Apps {
+		sem <- struct{}{}
 		wg.Add(1)
-		go r.validateChart(app, s, &wg, c)
+		go func(r *release, app string) {
+			defer func() {
+				wg.Done()
+				<-sem
+			}()
+			r.validateChart(app, s, c)
+		}(r, app)
 	}
 	wg.Wait()
 	close(c)
@@ -164,9 +172,8 @@ func validateReleaseCharts(s *state) error {
 
 var versionExtractor = regexp.MustCompile(`[\n]version:\s?(.*)`)
 
-func (r *release) validateChart(app string, s *state, wg *sync.WaitGroup, c chan string) {
+func (r *release) validateChart(app string, s *state, c chan string) {
 
-	defer wg.Done()
 	validateCurrentChart := true
 	if !r.isConsideredToRun(s) {
 		validateCurrentChart = false