fix: Fix needs to work for upgrades and when selectors are provided (#922)

* fix: Fix `needs` to work for upgrades and when selectors are provided

Fixes #919

* Add test framework for `helmfile apply`

* Various enhancements and fixes to the DAG support

- Make the order of upgrades/deletes more deterministic for testability
- Fix the test framework so that we can validate log outputs and errors
- Add more test cases for `helmfile apply`, along with bug fixes.
- Make sure it fails with an intuitive error when you have non-existent releases referenced from witin "needs"

pkg/app/app.go

@@ -1,6 +1,7 @@
 package app
 
 import (
+	"bytes"
 	"fmt"
 	"io/ioutil"
 	"log"
@@ -8,8 +9,10 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
+	"text/tabwriter"
 
 	"github.com/gosuri/uitable"
+	"github.com/roboll/helmfile/pkg/argparser"
 	"github.com/roboll/helmfile/pkg/helmexec"
 	"github.com/roboll/helmfile/pkg/remote"
 	"github.com/roboll/helmfile/pkg/state"
@@ -146,7 +149,7 @@ func (a *App) Sync(c SyncConfigProvider) error {
 
 func (a *App) Apply(c ApplyConfigProvider) error {
 	return a.ForEachState(func(run *Run) []error {
-		return run.Apply(c)
+		return a.apply(run, c)
 	})
 }
 
@@ -159,13 +162,13 @@ func (a *App) Status(c StatusesConfigProvider) error {
 func (a *App) Delete(c DeleteConfigProvider) error {
 	return a.reverse().ForEachState(func(run *Run) []error {
 		return run.Delete(c)
-	})
+	}, true)
 }
 
 func (a *App) Destroy(c DestroyConfigProvider) error {
 	return a.reverse().ForEachState(func(run *Run) []error {
 		return run.Destroy(c)
-	})
+	}, true)
 }
 
 func (a *App) Test(c TestConfigProvider) error {
@@ -380,8 +383,10 @@ func (a *App) visitStates(fileOrDir string, defOpts LoadOpts, converge func(*sta
 		if tmplErr != nil {
 			return appError(fmt.Sprintf("failed executing release templates in \"%s\"", f), tmplErr)
 		}
+
 		processed, errs := converge(templated, helm)
 		noMatchInHelmfiles = noMatchInHelmfiles && !processed
+
 		return context{a, templated}.clean(errs)
 	})
 
@@ -396,13 +401,13 @@ func (a *App) visitStates(fileOrDir string, defOpts LoadOpts, converge func(*sta
 	return nil
 }
 
-func (a *App) ForEachState(do func(*Run) []error) error {
+func (a *App) ForEachState(do func(*Run) []error, dagEnabled ...bool) error {
 	ctx := NewContext()
 	err := a.VisitDesiredStatesWithReleasesFiltered(a.FileOrDir, func(st *state.HelmState, helm helmexec.Interface) []error {
 		run := NewRun(st, helm, ctx)
 
 		return do(run)
-	})
+	}, dagEnabled...)
 
 	if err != nil && a.ErrorHandler != nil {
 		return a.ErrorHandler(err)
@@ -411,7 +416,74 @@ func (a *App) ForEachState(do func(*Run) []error) error {
 	return err
 }
 
-func (a *App) VisitDesiredStatesWithReleasesFiltered(fileOrDir string, converge func(*state.HelmState, helmexec.Interface) []error) error {
+func printBatches(batches [][]state.Release) string {
+	buf := &bytes.Buffer{}
+
+	w := new(tabwriter.Writer)
+
+	w.Init(buf, 0, 1, 1, ' ', 0)
+
+	fmt.Fprintln(w, "GROUP\tRELEASES")
+
+	for i, batch := range batches {
+		ids := []string{}
+		for _, r := range batch {
+			ids = append(ids, state.ReleaseToID(&r.ReleaseSpec))
+		}
+		fmt.Fprintf(w, "%d\t%s\n", i+1, strings.Join(ids, ", "))
+	}
+
+	w.Flush()
+
+	return buf.String()
+}
+
+func withDAG(templated *state.HelmState, helm helmexec.Interface, logger *zap.SugaredLogger, reverse bool, converge func(*state.HelmState, helmexec.Interface) (bool, []error)) (bool, []error) {
+	batches, err := state.PlanReleases(templated.Releases, templated.Selectors, reverse)
+	if err != nil {
+		return false, []error{err}
+	}
+
+	numBatches := len(batches)
+
+	logger.Debugf("processing %d groups of releases in this order:\n%s", numBatches, printBatches(batches))
+
+	all := true
+
+	for i, batch := range batches {
+		var targets []state.ReleaseSpec
+
+		for _, marked := range batch {
+			targets = append(targets, marked.ReleaseSpec)
+		}
+
+		var releaseIds []string
+		for _, r := range targets {
+			releaseIds = append(releaseIds, state.ReleaseToID(&r))
+		}
+
+		logger.Debugf("processing releases in group %d/%d: %s", i+1, numBatches, strings.Join(releaseIds, ", "))
+
+		batchSt := *templated
+		batchSt.Releases = targets
+
+		processed, errs := converge(&batchSt, helm)
+
+		if errs != nil && len(errs) > 0 {
+			return false, errs
+		}
+
+		all = all && processed
+	}
+
+	return all, nil
+}
+
+type Opts struct {
+	DAGEnabled bool
+}
+
+func (a *App) visitStatesWithSelectorsAndRemoteSupport(fileOrDir string, converge func(*state.HelmState, helmexec.Interface) (bool, []error)) error {
 	opts := LoadOpts{
 		Selectors: a.Selectors,
 	}
@@ -450,7 +522,11 @@ func (a *App) VisitDesiredStatesWithReleasesFiltered(fileOrDir string, converge
 
 	a.remote = remote
 
-	return a.visitStates(fileOrDir, opts, func(st *state.HelmState, helm helmexec.Interface) (bool, []error) {
+	return a.visitStates(fileOrDir, opts, converge)
+}
+
+func (a *App) Wrap(converge func(*state.HelmState, helmexec.Interface) []error) func(st *state.HelmState, helm helmexec.Interface) (bool, []error) {
+	return func(st *state.HelmState, helm helmexec.Interface) (bool, []error) {
 		if len(st.Selectors) > 0 {
 			err := st.FilterReleases()
 			if err != nil {
@@ -485,6 +561,20 @@ func (a *App) VisitDesiredStatesWithReleasesFiltered(fileOrDir string, converge
 		processed := len(st.Releases) != 0 && len(errs) == 0
 
 		return processed, errs
+	}
+}
+
+func (a *App) VisitDesiredStatesWithReleasesFiltered(fileOrDir string, converge func(*state.HelmState, helmexec.Interface) []error, dagEnabled ...bool) error {
+	f := a.Wrap(converge)
+
+	if len(dagEnabled) > 0 && dagEnabled[0] {
+		return a.visitStatesWithSelectorsAndRemoteSupport(fileOrDir, func(st *state.HelmState, helm helmexec.Interface) (bool, []error) {
+			return withDAG(st, helm, a.Logger, a.Reverse, f)
+		})
+	}
+
+	return a.visitStatesWithSelectorsAndRemoteSupport(fileOrDir, func(st *state.HelmState, helm helmexec.Interface) (bool, []error) {
+		return f(st, helm)
 	})
 }
 
@@ -560,6 +650,159 @@ func (a *App) findDesiredStateFiles(specifiedPath string) ([]string, error) {
 	return files, nil
 }
 
+func (a *App) apply(r *Run, c ApplyConfigProvider) []error {
+	st := r.state
+	helm := r.helm
+	ctx := r.ctx
+
+	affectedReleases := state.AffectedReleases{}
+	if !c.SkipDeps() {
+		if errs := ctx.SyncReposOnce(st, helm); errs != nil && len(errs) > 0 {
+			return errs
+		}
+		if errs := st.BuildDeps(helm); errs != nil && len(errs) > 0 {
+			return errs
+		}
+	}
+	if errs := st.PrepareReleases(helm, "apply"); errs != nil && len(errs) > 0 {
+		return errs
+	}
+
+	// helm must be 2.11+ and helm-diff should be provided `--detailed-exitcode` in order for `helmfile apply` to work properly
+	detailedExitCode := true
+
+	diffOpts := &state.DiffOpts{
+		NoColor: c.NoColor(),
+		Context: c.Context(),
+		Set:     c.Set(),
+	}
+
+	changedReleases, errs := st.DiffReleases(helm, c.Values(), c.Concurrency(), detailedExitCode, c.SuppressSecrets(), false, diffOpts)
+
+	deletingReleases, err := st.DetectReleasesToBeDeleted(helm, st.Releases)
+	if err != nil {
+		errs = append(errs, err)
+	}
+
+	fatalErrs := []error{}
+
+	noError := true
+	for _, e := range errs {
+		switch err := e.(type) {
+		case *state.ReleaseError:
+			if err.Code != 2 {
+				noError = false
+				fatalErrs = append(fatalErrs, e)
+			}
+		default:
+			noError = false
+			fatalErrs = append(fatalErrs, e)
+		}
+	}
+
+	releasesToBeDeleted := map[string]state.ReleaseSpec{}
+	for _, r := range deletingReleases {
+		id := state.ReleaseToID(&r)
+		releasesToBeDeleted[id] = r
+	}
+
+	releasesToBeUpdated := map[string]state.ReleaseSpec{}
+	for _, r := range changedReleases {
+		id := state.ReleaseToID(&r)
+
+		// If `helm-diff` detected changes but it is not being `helm delete`ed, we should run `helm upgrade`
+		if _, ok := releasesToBeDeleted[id]; !ok {
+			releasesToBeUpdated[id] = r
+		}
+	}
+
+	// sync only when there are changes
+	if noError {
+		if len(releasesToBeUpdated) == 0 && len(releasesToBeDeleted) == 0 {
+			// TODO better way to get the logger
+			logger := c.Logger()
+			logger.Infof("")
+			logger.Infof("No affected releases")
+		} else {
+			names := []string{}
+			for _, r := range releasesToBeUpdated {
+				names = append(names, fmt.Sprintf("  %s (%s) UPDATED", r.Name, r.Chart))
+			}
+			for _, r := range releasesToBeDeleted {
+				names = append(names, fmt.Sprintf("  %s (%s) DELETED", r.Name, r.Chart))
+			}
+			// Make the output deterministic for testing purpose
+			sort.Strings(names)
+
+			infoMsg := fmt.Sprintf(`Affected releases are:
+%s
+`, strings.Join(names, "\n"))
+			confMsg := fmt.Sprintf(`%s
+Do you really want to apply?
+  Helmfile will apply all your changes, as shown above.
+
+`, infoMsg)
+			interactive := c.Interactive()
+			if !interactive {
+				a.Logger.Debug(infoMsg)
+			}
+			if !interactive || interactive && r.askForConfirmation(confMsg) {
+				r.helm.SetExtraArgs(argparser.GetArgs(c.Args(), r.state)...)
+
+				// We deleted releases by traversing the DAG in reverse order
+				if len(releasesToBeDeleted) > 0 {
+					_, errs := withDAG(st, helm, a.Logger, true, a.Wrap(func(subst *state.HelmState, helm helmexec.Interface) []error {
+						var rs []state.ReleaseSpec
+
+						for _, r := range subst.Releases {
+							if _, ok := releasesToBeDeleted[state.ReleaseToID(&r)]; ok {
+								rs = append(rs, r)
+							}
+						}
+
+						subst.Releases = rs
+
+						return subst.DeleteReleasesForSync(&affectedReleases, helm, c.Concurrency())
+					}))
+
+					if errs != nil && len(errs) > 0 {
+						return errs
+					}
+				}
+
+				// We upgrade releases by traversing the DAG
+				if len(releasesToBeUpdated) > 0 {
+					_, errs := withDAG(st, helm, a.Logger, false, a.Wrap(func(subst *state.HelmState, helm helmexec.Interface) []error {
+						var rs []state.ReleaseSpec
+
+						for _, r := range subst.Releases {
+							if _, ok := releasesToBeUpdated[state.ReleaseToID(&r)]; ok {
+								rs = append(rs, r)
+							}
+						}
+
+						subst.Releases = rs
+
+						syncOpts := state.SyncOpts{
+							Set: c.Set(),
+						}
+						return subst.SyncReleases(&affectedReleases, helm, c.Values(), c.Concurrency(), &syncOpts)
+					}))
+
+					if errs != nil && len(errs) > 0 {
+						return errs
+					}
+				}
+
+				return nil
+			}
+		}
+	}
+
+	affectedReleases.DisplayAffectedReleases(c.Logger())
+	return fatalErrs
+}
+
 func fileExistsAt(path string) bool {
 	fileInfo, err := os.Stat(path)
 	return err == nil && fileInfo.Mode().IsRegular()

pkg/app/desired_state_file_loader.go

@@ -76,10 +76,10 @@ func (ld *desiredStateLoader) Load(f string, opts LoadOpts) (*state.HelmState, e
 	}
 
 	if ld.namespace != "" {
-		if st.Namespace != "" {
+		if st.OverrideNamespace != "" {
 			return nil, errors.New("err: Cannot use option --namespace and set attribute namespace.")
 		}
-		st.Namespace = ld.namespace
+		st.OverrideNamespace = ld.namespace
 	}
 
 	return st, nil

pkg/app/run.go

@@ -118,104 +118,6 @@ Do you really want to delete?
 	return errs
 }
 
-func (r *Run) Apply(c ApplyConfigProvider) []error {
-	st := r.state
-	helm := r.helm
-	ctx := r.ctx
-
-	affectedReleases := state.AffectedReleases{}
-	if !c.SkipDeps() {
-		if errs := ctx.SyncReposOnce(st, helm); errs != nil && len(errs) > 0 {
-			return errs
-		}
-		if errs := st.BuildDeps(helm); errs != nil && len(errs) > 0 {
-			return errs
-		}
-	}
-	if errs := st.PrepareReleases(helm, "apply"); errs != nil && len(errs) > 0 {
-		return errs
-	}
-
-	// helm must be 2.11+ and helm-diff should be provided `--detailed-exitcode` in order for `helmfile apply` to work properly
-	detailedExitCode := true
-
-	diffOpts := &state.DiffOpts{
-		NoColor: c.NoColor(),
-		Context: c.Context(),
-		Set:     c.Set(),
-	}
-
-	releases, errs := st.DiffReleases(helm, c.Values(), c.Concurrency(), detailedExitCode, c.SuppressSecrets(), false, diffOpts)
-
-	releasesToBeDeleted, err := st.DetectReleasesToBeDeleted(helm)
-	if err != nil {
-		errs = append(errs, err)
-	}
-
-	fatalErrs := []error{}
-
-	noError := true
-	for _, e := range errs {
-		switch err := e.(type) {
-		case *state.ReleaseError:
-			if err.Code != 2 {
-				noError = false
-				fatalErrs = append(fatalErrs, e)
-			}
-		default:
-			noError = false
-			fatalErrs = append(fatalErrs, e)
-		}
-	}
-
-	// sync only when there are changes
-	if noError {
-		if len(releases) == 0 && len(releasesToBeDeleted) == 0 {
-			// TODO better way to get the logger
-			logger := c.Logger()
-			logger.Infof("")
-			logger.Infof("No affected releases")
-		} else {
-			names := []string{}
-			for _, r := range releases {
-				names = append(names, fmt.Sprintf("  %s (%s) UPDATED", r.Name, r.Chart))
-			}
-			for _, r := range releasesToBeDeleted {
-				names = append(names, fmt.Sprintf("  %s (%s) DELETED", r.Name, r.Chart))
-			}
-
-			msg := fmt.Sprintf(`Affected releases are:
-%s
-
-Do you really want to apply?
-  Helmfile will apply all your changes, as shown above.
-
-`, strings.Join(names, "\n"))
-			interactive := c.Interactive()
-			if !interactive || interactive && r.askForConfirmation(msg) {
-				rs := []state.ReleaseSpec{}
-				for _, r := range releases {
-					rs = append(rs, *r)
-				}
-				for _, r := range releasesToBeDeleted {
-					rs = append(rs, *r)
-				}
-
-				r.helm.SetExtraArgs(argparser.GetArgs(c.Args(), r.state)...)
-
-				st.Releases = rs
-				syncOpts := &state.SyncOpts{
-					Set: c.Set(),
-				}
-				return st.SyncReleases(&affectedReleases, helm, c.Values(), c.Concurrency(), syncOpts)
-			}
-		}
-	}
-
-	affectedReleases.DisplayAffectedReleases(c.Logger())
-	return fatalErrs
-}
-
 func (r *Run) Diff(c DiffConfigProvider) []error {
 	st := r.state
 	helm := r.helm

pkg/app/two_pass_renderer.go

@@ -31,12 +31,15 @@ func (r *desiredStateLoader) renderPrestate(firstPassEnv *environment.Environmen
 	yamlBuf, err := firstPassRenderer.RenderTemplateContentToBuffer(content)
 	if err != nil && r.logger != nil {
 		r.logger.Debugf("first-pass rendering input of \"%s\":\n%s", filename, prependLineNumbers(string(content)))
+		r.logger.Debugf("template syntax error: %v", err)
 		if yamlBuf == nil { // we have a template syntax error, let the second parse report
-			r.logger.Debugf("template syntax error: %v", err)
 			return firstPassEnv, nil
 		}
 	}
 	yamlData := yamlBuf.String()
+	if r.logger != nil {
+		r.logger.Debugf("first-pass rendering output of \"%s\":\n%s", filename, prependLineNumbers(yamlData))
+	}
 
 	// Work-around for https://github.com/golang/go/issues/24963
 	sanitized := strings.ReplaceAll(yamlData, "<no value>", "")