Handle environment secrets concurrently

Ref #782

pkg/state/create.go

@@ -4,14 +4,15 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"io"
+	"os"
+
 	"github.com/imdario/mergo"
 	"github.com/roboll/helmfile/pkg/environment"
 	"github.com/roboll/helmfile/pkg/helmexec"
 	"github.com/roboll/helmfile/pkg/maputil"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v2"
-	"io"
-	"os"
 )
 
 type StateLoadError struct {
@@ -201,59 +202,103 @@ func (st *HelmState) loadEnvValues(name string, ctxEnv *environment.Environment,
 
 				envSecretFiles = append(envSecretFiles, resolved...)
 			}
+			if err = st.scatterGatherEnvSecretFiles(envSecretFiles, helm, envVals, readFile); err != nil {
+				return nil, err
+			}
+		}
+	} else if ctxEnv == nil && name != DefaultEnv {
+		return nil, &UndefinedEnvError{msg: fmt.Sprintf("environment \"%s\" is not defined", name)}
+	}
+
+	newEnv := &environment.Environment{Name: name, Values: envVals}
+
+	if ctxEnv != nil {
+		intEnv := *ctxEnv
+
+		if err := mergo.Merge(&intEnv, newEnv, mergo.WithOverride); err != nil {
+			return nil, fmt.Errorf("error while merging environment values for \"%s\": %v", name, err)
+		}
+
+		newEnv = &intEnv
+	}
+
+	return newEnv, nil
+}
+
+func (st *HelmState) scatterGatherEnvSecretFiles(envSecretFiles []string, helm helmexec.Interface, envVals map[string]interface{}, readFile func(string) ([]byte, error)) error {
+	var errs []error
+
+	inputs := envSecretFiles
+	inputsSize := len(inputs)
 
-			for _, path := range envSecretFiles {
-				// Work-around to allow decrypting environment secrets
-				//
-				// We don't have releases loaded yet and therefore unable to decide whether
-				// helmfile should use helm-tiller to call helm-secrets or not.
-				//
-				// This means that, when you use environment secrets + tillerless setup, you still need a tiller
-				// installed on the cluster, just for decrypting secrets!
-				// Related: https://github.com/futuresimple/helm-secrets/issues/83
+	type secretResult struct {
+		result map[string]interface{}
+		err    error
+		path   string
+	}
+
+	secrets := make(chan string, inputsSize)
+	results := make(chan secretResult, inputsSize)
+
+	st.scatterGather(0, inputsSize,
+		func() {
+			for _, secretFile := range envSecretFiles {
+				secrets <- secretFile
+			}
+			close(secrets)
+		},
+		func(id int) {
+			for path := range secrets {
 				release := &ReleaseSpec{}
 				flags := st.appendConnectionFlags([]string{}, release)
 				decFile, err := helm.DecryptSecret(st.createHelmContext(release, 0), path, flags...)
 				if err != nil {
-					return nil, err
+					results <- secretResult{nil, err, path}
+					continue
 				}
 				bytes, err := readFile(decFile)
 				if err != nil {
-					return nil, fmt.Errorf("failed to load environment secrets file \"%s\": %v", path, err)
+					results <- secretResult{nil, fmt.Errorf("failed to load environment secrets file \"%s\": %v", path, err), path}
+					continue
 				}
 				m := map[string]interface{}{}
 				if err := yaml.Unmarshal(bytes, &m); err != nil {
-					return nil, fmt.Errorf("failed to load environment secrets file \"%s\": %v", path, err)
+					results <- secretResult{nil, fmt.Errorf("failed to load environment secrets file \"%s\": %v", path, err), path}
+					continue
 				}
 				// All the nested map key should be string. Otherwise we get strange errors due to that
 				// mergo or reflect is unable to merge map[interface{}]interface{} with map[string]interface{} or vice versa.
 				// See https://github.com/roboll/helmfile/issues/677
 				vals, err := maputil.CastKeysToStrings(m)
 				if err != nil {
-					return nil, err
+					results <- secretResult{nil, fmt.Errorf("failed to load environment secrets file \"%s\": %v", path, err), path}
+					continue
 				}
-				if err := mergo.Merge(&envVals, &vals, mergo.WithOverride); err != nil {
-					return nil, fmt.Errorf("failed to load \"%s\": %v", path, err)
+				results <- secretResult{vals, nil, path}
+			}
+		},
+		func() {
+			for i := 0; i < inputsSize; i++ {
+				result := <-results
+				if result.err != nil {
+					errs = append(errs, result.err)
+				} else {
+					if err := mergo.Merge(&envVals, &result.result, mergo.WithOverride); err != nil {
+						errs = append(errs, fmt.Errorf("failed to load environment secrets file \"%s\": %v", result.path, err))
+					}
 				}
 			}
-		}
-	} else if ctxEnv == nil && name != DefaultEnv {
-		return nil, &UndefinedEnvError{msg: fmt.Sprintf("environment \"%s\" is not defined", name)}
-	}
-
-	newEnv := &environment.Environment{Name: name, Values: envVals}
-
-	if ctxEnv != nil {
-		intEnv := *ctxEnv
+			close(results)
+		},
+	)
 
-		if err := mergo.Merge(&intEnv, newEnv, mergo.WithOverride); err != nil {
-			return nil, fmt.Errorf("error while merging environment values for \"%s\": %v", name, err)
+	if len(errs) > 1 {
+		for _, err := range errs {
+			st.logger.Error(err)
 		}
-
-		newEnv = &intEnv
+		return fmt.Errorf("Failed loading environment secrets with %d errors", len(errs))
 	}
-
-	return newEnv, nil
+	return nil
 }
 
 func (st *HelmState) loadValuesEntries(missingFileHandler *string, entries []interface{}) (map[string]interface{}, error) {