Repository: grafana/tanka
Pull request #473 (2021-01-08 12:15:49)

refactor: parallel
As suggested in #471, this refactors the parallel functions to use the chan idioms. It also makes `parallelOpts` a
private struct.


Patch:
diff --git a/cmd/tk/export.go b/cmd/tk/export.go
index babf5631..dddc7705 100644
--- a/cmd/tk/export.go
+++ b/cmd/tk/export.go
@@ -34,6 +34,7 @@ func exportCmd() *cli.Command {
 
 	extension := cmd.Flags().String("extension", "yaml", "File extension")
 	merge := cmd.Flags().Bool("merge", false, "Allow merging with existing directory")
+	parallel := cmd.Flags().IntP("parallel", "p", 8, "Number of environments to process in parallel")
 
 	vars := workflowFlags(cmd.Flags())
 	getJsonnetOpts := jsonnetFlags(cmd.Flags())
@@ -47,10 +48,11 @@ func exportCmd() *cli.Command {
 			Extension: *extension,
 			Merge:     *merge,
 			Targets:   vars.targets,
-			ParallelOpts: tanka.ParallelOpts{
+			Opts: tanka.Opts{
 				JsonnetOpts: getJsonnetOpts(),
-				Selector:    getLabelSelector(),
 			},
+			Selector:    getLabelSelector(),
+			Parallelism: *parallel,
 		}
 
 		var paths []string
@@ -63,7 +65,7 @@ func exportCmd() *cli.Command {
 				}
 
 				// get absolute path to Environment
-				envs, err := tanka.ListEnvs(path, tanka.ListOpts{Selector: opts.ParallelOpts.Selector})
+				envs, err := tanka.ListEnvs(path, tanka.ListOpts{Selector: opts.Selector})
 				if err != nil {
 					return err
 				}
@@ -75,7 +77,7 @@ func exportCmd() *cli.Command {
 			}
 
 			// validate environment
-			if _, err := tanka.Peek(path, opts.ParallelOpts.JsonnetOpts); err != nil {
+			if _, err := tanka.Peek(path, opts.Opts.JsonnetOpts); err != nil {
 				return err
 			}
 
diff --git a/pkg/spec/spec.go b/pkg/spec/spec.go
index 42ae491f..2c85ba06 100644
--- a/pkg/spec/spec.go
+++ b/pkg/spec/spec.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/pkg/errors"
 
+	"github.com/grafana/tanka/pkg/jsonnet/jpath"
 	"github.com/grafana/tanka/pkg/spec/v1alpha1"
 )
 
@@ -20,30 +21,43 @@ const Specfile = "spec.json"
 
 // ParseDir parses the given environments `spec.json` into a `v1alpha1.Environment`
 // object with the name set to the directories name
-func ParseDir(baseDir, path string) (*v1alpha1.Environment, error) {
-	fi, err := os.Stat(baseDir)
+func ParseDir(path string) (*v1alpha1.Environment, error) {
+	root, base, err := jpath.Dirs(path)
 	if err != nil {
 		return nil, err
 	}
-	if !fi.IsDir() {
-		return nil, errors.New("baseDir is not an directory")
+
+	// name of the environment: relative path from rootDir
+	name, err := filepath.Rel(root, base)
+	if err != nil {
+		return nil, err
+	}
+
+	file, err := jpath.Entrypoint(path)
+	if err != nil {
+		return nil, err
+	}
+
+	namespace, err := filepath.Rel(root, file)
+	if err != nil {
+		return nil, err
 	}
 
-	data, err := ioutil.ReadFile(filepath.Join(baseDir, Specfile))
+	data, err := ioutil.ReadFile(filepath.Join(base, Specfile))
 	if err != nil {
 		if os.IsNotExist(err) {
 			c := v1alpha1.New()
-			c.Metadata.Name = path // legacy behavior
-			c.Metadata.Namespace = path
+			c.Metadata.Name = name // legacy behavior
+			c.Metadata.Namespace = namespace
 			return c, ErrNoSpec{path}
 		}
 		return nil, err
 	}
 
-	c, err := Parse(data, path)
+	c, err := Parse(data, namespace)
 	if c != nil {
 		// set the name field
-		c.Metadata.Name = path // legacy behavior
+		c.Metadata.Name = name // legacy behavior
 	}
 
 	return c, err
diff --git a/pkg/tanka/export.go b/pkg/tanka/export.go
index 73eca513..970577a8 100644
--- a/pkg/tanka/export.go
+++ b/pkg/tanka/export.go
@@ -12,6 +12,7 @@ import (
 	"text/template"
 
 	"github.com/Masterminds/sprig/v3"
+	"k8s.io/apimachinery/pkg/labels"
 
 	"github.com/grafana/tanka/pkg/kubernetes/manifest"
 	"github.com/grafana/tanka/pkg/process"
@@ -37,8 +38,12 @@ type ExportEnvOpts struct {
 	Merge bool
 	// optional: only export specified Kubernetes manifests
 	Targets []string
-	// optional: options for parsing Environments
-	ParallelOpts ParallelOpts
+	// optional: options to parse Jsonnet
+	Opts Opts
+	// optional: filter environments based on labels
+	Selector labels.Selector
+	// optional: number of environments to process in parallel
+	Parallelism int
 }
 
 func ExportEnvironments(paths []string, to string, opts *ExportEnvOpts) error {
@@ -55,7 +60,11 @@ func ExportEnvironments(paths []string, to string, opts *ExportEnvOpts) error {
 	}
 
 	// get all environments for paths
-	envs, err := parallelLoadEnvironments(paths, opts.ParallelOpts)
+	envs, err := parallelLoadEnvironments(paths, parallelOpts{
+		JsonnetOpts: opts.Opts.JsonnetOpts,
+		Selector:    opts.Selector,
+		Parallelism: opts.Parallelism,
+	})
 	if err != nil {
 		return err
 	}
diff --git a/pkg/tanka/inline.go b/pkg/tanka/inline.go
index b418157d..4137b4c6 100644
--- a/pkg/tanka/inline.go
+++ b/pkg/tanka/inline.go
@@ -45,12 +45,17 @@ func (i *InlineLoader) Load(path string, opts JsonnetOpts) (*v1alpha1.Environmen
 		return nil, fmt.Errorf("Found no environments in '%s'", path)
 	}
 
-	root, base, err := jpath.Dirs(path)
+	root, err := jpath.FindRoot(path)
 	if err != nil {
 		return nil, err
 	}
 
-	name, err := filepath.Rel(root, base)
+	file, err := jpath.Entrypoint(path)
+	if err != nil {
+		return nil, err
+	}
+
+	namespace, err := filepath.Rel(root, file)
 	if err != nil {
 		return nil, err
 	}
@@ -61,7 +66,7 @@ func (i *InlineLoader) Load(path string, opts JsonnetOpts) (*v1alpha1.Environmen
 		return nil, err
 	}
 
-	env, err := spec.Parse(envData, name)
+	env, err := spec.Parse(envData, namespace)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/tanka/parallel.go b/pkg/tanka/parallel.go
index e3f4bfb4..e623536c 100644
--- a/pkg/tanka/parallel.go
+++ b/pkg/tanka/parallel.go
@@ -2,59 +2,56 @@ package tanka
 
 import (
 	"fmt"
-	"sync"
+
+	"k8s.io/apimachinery/pkg/labels"
 
 	"github.com/grafana/tanka/pkg/spec/v1alpha1"
 )
 
-const parallel = 8
+const defaultParallelism = 8
+
+type parallelOpts struct {
+	JsonnetOpts JsonnetOpts
+	Selector    labels.Selector
+	Parallelism int
+}
 
 // parallelLoadEnvironments evaluates multiple environments in parallel
-func parallelLoadEnvironments(paths []string, opts ParallelOpts) (envs []*v1alpha1.Environment, err error) {
-	wg := sync.WaitGroup{}
-	envsChan := make(chan parallelJob)
-	var allErrors []error
+func parallelLoadEnvironments(paths []string, opts parallelOpts) ([]*v1alpha1.Environment, error) {
+	jobsCh := make(chan parallelJob)
+	outCh := make(chan parallelOut)
 
-	numParallel := parallel
-	if opts.Parallel > 0 {
-		numParallel = opts.Parallel
-	}
-	for i := 0; i < numParallel; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			errs := parallelWorker(envsChan)
-			if errs != nil {
-				allErrors = append(allErrors, errs...)
-			}
-		}()
+	if opts.Parallelism <= 0 {
+		opts.Parallelism = defaultParallelism
 	}
 
-	results := make([]*v1alpha1.Environment, 0, len(paths))
+	for i := 0; i < opts.Parallelism; i++ {
+		go parallelWorker(jobsCh, outCh)
+	}
 
 	for _, path := range paths {
-		env := &v1alpha1.Environment{}
-		results = append(results, env)
-		envsChan <- parallelJob{
+		jobsCh <- parallelJob{
 			path: path,
-			opts: opts.JsonnetOpts,
-			env:  env,
+			opts: Opts{JsonnetOpts: opts.JsonnetOpts},
 		}
 	}
-	close(envsChan)
-	wg.Wait()
+	close(jobsCh)
 
-	for _, env := range results {
-		if env == nil {
+	var envs []*v1alpha1.Environment
+	var errors []error
+	for i := 0; i < len(paths); i++ {
+		out := <-outCh
+		if out.err != nil {
+			errors = append(errors, out.err)
 			continue
 		}
-		if opts.Selector == nil || opts.Selector.Empty() || opts.Selector.Matches(env.Metadata) {
-			envs = append(envs, env)
+		if opts.Selector == nil || opts.Selector.Empty() || opts.Selector.Matches(out.env.Metadata) {
+			envs = append(envs, out.env)
 		}
 	}
 
-	if len(allErrors) != 0 {
-		return envs, ErrParallel{errors: allErrors}
+	if len(errors) != 0 {
+		return envs, ErrParallel{errors: errors}
 	}
 
 	return envs, nil
@@ -62,21 +59,20 @@ func parallelLoadEnvironments(paths []string, opts ParallelOpts) (envs []*v1alph
 
 type parallelJob struct {
 	path string
-	opts JsonnetOpts
-	env  *v1alpha1.Environment
+	opts Opts
 }
 
-func parallelWorker(envsChan <-chan parallelJob) (errs []error) {
-	for req := range envsChan {
-		env, err := LoadEnvironment(req.path, Opts{JsonnetOpts: req.opts})
+type parallelOut struct {
+	env *v1alpha1.Environment
+	err error
+}
+
+func parallelWorker(jobsCh <-chan parallelJob, outCh chan parallelOut) {
+	for job := range jobsCh {
+		env, err := LoadEnvironment(job.path, job.opts)
 		if err != nil {
-			errs = append(errs, fmt.Errorf("%s:\n %w", req.path, err))
-			continue
+			err = fmt.Errorf("%s:\n %w", job.path, err)
 		}
-		*req.env = *env
-	}
-	if len(errs) != 0 {
-		return errs
+		outCh <- parallelOut{env: env, err: err}
 	}
-	return nil
 }
diff --git a/pkg/tanka/static.go b/pkg/tanka/static.go
index 72ca4e57..445ca0f9 100644
--- a/pkg/tanka/static.go
+++ b/pkg/tanka/static.go
@@ -3,9 +3,7 @@ package tanka
 import (
 	"encoding/json"
 	"log"
-	"path/filepath"
 
-	"github.com/grafana/tanka/pkg/jsonnet/jpath"
 	"github.com/grafana/tanka/pkg/spec"
 	"github.com/grafana/tanka/pkg/spec/v1alpha1"
 )
@@ -33,12 +31,7 @@ func (s StaticLoader) Load(path string, opts JsonnetOpts) (*v1alpha1.Environment
 }
 
 func (s StaticLoader) Peek(path string, opts JsonnetOpts) (*v1alpha1.Environment, error) {
-	root, base, err := jpath.Dirs(path)
-	if err != nil {
-		return nil, err
-	}
-
-	config, err := parseStaticSpec(root, base)
+	config, err := parseStaticSpec(path)
 	if err != nil {
 		return nil, err
 	}
@@ -48,14 +41,8 @@ func (s StaticLoader) Peek(path string, opts JsonnetOpts) (*v1alpha1.Environment
 
 // parseStaticSpec parses the `spec.json` of the environment and returns a
 // *kubernetes.Kubernetes from it
-func parseStaticSpec(root, base string) (*v1alpha1.Environment, error) {
-	// name of the environment: relative path from rootDir
-	name, err := filepath.Rel(root, base)
-	if err != nil {
-		return nil, err
-	}
-
-	env, err := spec.ParseDir(base, name)
+func parseStaticSpec(path string) (*v1alpha1.Environment, error) {
+	env, err := spec.ParseDir(path)
 	if err != nil {
 		switch err.(type) {
 		// the config includes deprecated fields
diff --git a/pkg/tanka/tanka.go b/pkg/tanka/tanka.go
index 710ecccc..36ce6975 100644
--- a/pkg/tanka/tanka.go
+++ b/pkg/tanka/tanka.go
@@ -8,7 +8,6 @@ import (
 	"fmt"
 
 	"github.com/Masterminds/semver"
-	"k8s.io/apimachinery/pkg/labels"
 
 	"github.com/grafana/tanka/pkg/jsonnet"
 	"github.com/grafana/tanka/pkg/process"
@@ -55,9 +54,3 @@ func checkVersion(constraint string) error {
 
 	return nil
 }
-
-type ParallelOpts struct {
-	JsonnetOpts JsonnetOpts
-	Selector    labels.Selector
-	Parallel    int
-}


Test patch:
diff --git a/pkg/tanka/load_test.go b/pkg/tanka/load_test.go
index 36084d61..0eab597b 100644
--- a/pkg/tanka/load_test.go
+++ b/pkg/tanka/load_test.go
@@ -32,7 +32,7 @@ func TestLoad(t *testing.T) {
 				Kind:       v1alpha1.New().Kind,
 				Metadata: v1alpha1.Metadata{
 					Name:      "cases/withspecjson",
-					Namespace: "cases/withspecjson",
+					Namespace: "cases/withspecjson/main.jsonnet",
 					Labels:    v1alpha1.New().Metadata.Labels,
 				},
 				Spec: v1alpha1.Spec{
@@ -62,7 +62,7 @@ func TestLoad(t *testing.T) {
 				Kind:       v1alpha1.New().Kind,
 				Metadata: v1alpha1.Metadata{
 					Name:      "cases/withspecjson",
-					Namespace: "cases/withspecjson",
+					Namespace: "cases/withspecjson/main.jsonnet",
 					Labels:    v1alpha1.New().Metadata.Labels,
 				},
 				Spec: v1alpha1.Spec{
@@ -79,6 +79,36 @@ func TestLoad(t *testing.T) {
 
 		{
 			name:    "inline",
+			baseDir: "./testdata/cases/withenv/",
+			expected: manifest.List{{
+				"apiVersion": "v1",
+				"kind":       "ConfigMap",
+				"metadata": map[string]interface{}{
+					"name":      "config",
+					"namespace": "withenv",
+				},
+			}},
+			env: &v1alpha1.Environment{
+				APIVersion: v1alpha1.New().APIVersion,
+				Kind:       v1alpha1.New().Kind,
+				Metadata: v1alpha1.Metadata{
+					Name:      "withenv",
+					Namespace: "cases/withenv/main.jsonnet",
+					Labels:    v1alpha1.New().Metadata.Labels,
+				},
+				Spec: v1alpha1.Spec{
+					APIServer: "https://localhost",
+					Namespace: "withenv",
+				},
+				Data: map[string]interface{}{
+					"apiVersion": "v1",
+					"kind":       "ConfigMap",
+					"metadata":   map[string]interface{}{"name": "config", "namespace": "withenv"},
+				},
+			},
+		},
+		{
+			name:    "inline-filename",
 			baseDir: "./testdata/cases/withenv/main.jsonnet",
 			expected: manifest.List{{
 				"apiVersion": "v1",
@@ -93,7 +123,7 @@ func TestLoad(t *testing.T) {
 				Kind:       v1alpha1.New().Kind,
 				Metadata: v1alpha1.Metadata{
 					Name:      "withenv",
-					Namespace: "cases/withenv",
+					Namespace: "cases/withenv/main.jsonnet",
 					Labels:    v1alpha1.New().Metadata.Labels,
 				},
 				Spec: v1alpha1.Spec{