Merge pull request #431 from Praqma/parallelise_releases

feat: Parallelise releases with the same priority

docs/cmd_reference.md

@@ -12,20 +12,23 @@ This lists available CMD options in Helmsman:
         apply the plan directly.
 
   `--context-override string`
-        override releases context defined in release state with this one.       
+        override releases context defined in release state with this one.
 
   `--debug`
         show the debug execution logs and actual helm/kubectl commands. This can log secrets and should only be used for debugging purposes.
 
   `--verbose`
-        show verbose execution logs.   
+        show verbose execution logs.
 
   `--destroy`
         delete all deployed releases.
 
   `--diff-context num`
         number of lines of context to show around changes in helm diff output.
 
+  `-p`
+        max number of concurrent helm releases to run
+
   `--dry-run`
         apply the dry-run (do not update) option for helm commands.
 
@@ -45,7 +48,7 @@ This lists available CMD options in Helmsman:
         path to the kubeconfig file to use for CLI requests.
 
   `--migrate-context`
-        Updates the context name for all apps defined in the DSF and applies Helmsman labels. Using this flag is required if you want to change context name after it has been set.      
+        Updates the context name for all apps defined in the DSF and applies Helmsman labels. Using this flag is required if you want to change context name after it has been set.
 
   `--no-banner`
         don't show the banner.

internal/app/cli.go

@@ -64,6 +64,7 @@ type cli struct {
 	version               bool
 	noCleanup             bool
 	migrateContext        bool
+	parallel              int
 }
 
 func printUsage() {
@@ -83,6 +84,7 @@ func (c *cli) parse() {
 	flag.Var(&c.target, "target", "limit execution to specific app.")
 	flag.Var(&c.group, "group", "limit execution to specific group of apps.")
 	flag.IntVar(&c.diffContext, "diff-context", -1, "number of lines of context to show around changes in helm diff output")
+	flag.IntVar(&c.parallel, "p", 1, "max number of concurrent helm releases to run")
 	flag.StringVar(&c.kubeconfig, "kubeconfig", "", "path to the kubeconfig file to use for CLI requests")
 	flag.StringVar(&c.nsOverride, "ns-override", "", "override defined namespaces with this one")
 	flag.StringVar(&c.contextOverride, "context-override", "", "override releases context defined in release state with this one")
@@ -138,6 +140,10 @@ func (c *cli) parse() {
 		log.Fatal("--target and --group can't be used together.")
 	}
 
+	if c.parallel < 1 {
+		c.parallel = 1
+	}
+
 	helmVersion := strings.TrimSpace(getHelmVersion())
 	extractedHelmVersion := helmVersion
 	if !strings.HasPrefix(helmVersion, "v") {

internal/app/decision_maker.go

@@ -170,14 +170,15 @@ var releaseNameExtractor = regexp.MustCompile(`sh\.helm\.release\.v\d+\.`)
 // The releases are categorized by the namespaces in which they are deployed
 // The returned map format is: map[<namespace>:map[<helmRelease>:true]]
 func (cs *currentState) getHelmsmanReleases(s *state) map[string]map[string]bool {
+	const outputFmt = "custom-columns=NAME:.metadata.name,CTX:.metadata.labels.HELMSMAN_CONTEXT"
 	var (
-		wg    sync.WaitGroup
-		mutex = &sync.Mutex{}
+		wg         sync.WaitGroup
+		mutex      = &sync.Mutex{}
+		namespaces map[string]namespace
 	)
-	const outputFmt = "custom-columns=NAME:.metadata.name,CTX:.metadata.labels.HELMSMAN_CONTEXT"
 	releases := make(map[string]map[string]bool)
 	sem := make(chan struct{}, resourcePool)
-	namespaces := make(map[string]namespace)
+
 	if len(s.TargetMap) > 0 {
 		namespaces = s.TargetNamespaces
 	} else {

internal/app/plan.go

@@ -86,25 +86,60 @@ func (p *plan) exec() {
 		log.Info("Nothing to execute")
 	}
 
+	wg := sync.WaitGroup{}
+	sem := make(chan struct{}, flags.parallel)
+	var fail bool
+	var priorities []int
+	pl := make(map[int][]orderedCommand)
 	for _, cmd := range p.Commands {
-		log.Notice(cmd.Command.Description)
-		result := cmd.Command.exec()
-		if cmd.targetRelease != nil && !flags.dryRun && !flags.destroy {
-			cmd.targetRelease.label()
+		pl[cmd.Priority] = append(pl[cmd.Priority], cmd)
+	}
+	for priority := range pl {
+		priorities = append(priorities, priority)
+	}
+	sort.Ints(priorities)
+
+	for _, priority := range priorities {
+		c := make(chan string, len(pl[priority]))
+		for _, cmd := range pl[priority] {
+			sem <- struct{}{}
+			wg.Add(1)
+			go func(cmd orderedCommand) {
+				defer func() {
+					wg.Done()
+					<-sem
+				}()
+				log.Notice(cmd.Command.Description)
+				result := cmd.Command.exec()
+				if cmd.targetRelease != nil && !flags.dryRun && !flags.destroy {
+					cmd.targetRelease.label()
+				}
+				if result.code != 0 {
+					errorMsg := result.errors
+					if !flags.verbose {
+						errorMsg = strings.Split(result.errors, "---")[0]
+					}
+					c <- fmt.Sprintf("Command for release [%s] returned [ %d ] exit code and error message [ %s ]", cmd.targetRelease.Name, result.code, strings.TrimSpace(errorMsg))
+				} else {
+					log.Notice(result.output)
+					log.Notice("Finished: " + cmd.Command.Description)
+					if _, err := url.ParseRequestURI(settings.SlackWebhook); err == nil {
+						notifySlack(cmd.Command.Description+" ... SUCCESS!", settings.SlackWebhook, false, true)
+					}
+				}
+			}(cmd)
 		}
-		if result.code != 0 {
-			errorMsg := result.errors
-			if !flags.verbose {
-				errorMsg = strings.Split(result.errors, "---")[0]
-			}
-			log.Fatal(fmt.Sprintf("Command returned [ %d ] exit code and error message [ %s ]", result.code, strings.TrimSpace(errorMsg)))
-		} else {
-			log.Notice(result.output)
-			log.Notice("Finished: " + cmd.Command.Description)
-			if _, err := url.ParseRequestURI(settings.SlackWebhook); err == nil {
-				notifySlack(cmd.Command.Description+" ... SUCCESS!", settings.SlackWebhook, false, true)
+		wg.Wait()
+		close(c)
+		for err := range c {
+			if err != "" {
+				fail = true
+				log.Error(err)
 			}
 		}
+		if fail {
+			log.Fatal("Plan execution failed")
+		}
 	}
 
 	if len(p.Commands) > 0 {