Change mechanism for component evaluation

internal/commands/common.go

@@ -92,6 +92,7 @@ type StdOptions interface {
 	Stdout() io.Writer                                     // output to write to
 	DefaultNamespace(env string) string                    // the default namespace for the supplied environment
 	Confirm(context string) error                          // confirmation function for dangerous operations
+	EvalConcurrency() int                                  // the concurrency using which to evaluate components
 }
 
 // Client encapsulates all remote operations needed for the superset of all commands.

internal/commands/filter.go

@@ -65,10 +65,11 @@ func filteredObjects(req StdOptions, env string, fp filterParams) ([]model.K8sLo
 	}
 	jvm := req.VM()
 	output, err := eval.Components(components, eval.Context{
-		App:     req.App().Name(),
-		Env:     env,
-		VM:      jvm,
-		Verbose: req.Verbosity() > 1,
+		App:         req.App().Name(),
+		Env:         env,
+		VM:          jvm,
+		Verbose:     req.Verbosity() > 1,
+		Concurrency: req.EvalConcurrency(),
 	})
 	if err != nil {
 		return nil, err

internal/commands/utils_test.go

@@ -101,12 +101,13 @@ func (c *client) Delete(obj model.K8sMeta, dryRun bool) (*remote.SyncResult, err
 }
 
 type opts struct {
-	app       *model.App
-	client    *client
-	colorize  bool
-	verbosity int
-	out       io.Writer
-	defaultNs string
+	app         *model.App
+	client      *client
+	colorize    bool
+	verbosity   int
+	out         io.Writer
+	defaultNs   string
+	concurrency int
 }
 
 func (o *opts) App() *model.App {
@@ -144,6 +145,10 @@ func (o *opts) Client(env string) (Client, error) {
 	return o.client, nil
 }
 
+func (o *opts) EvalConcurrency() int {
+	return o.concurrency
+}
+
 func (o *opts) Stdout() io.Writer {
 	return o.out
 }

internal/eval/eval.go

@@ -20,20 +20,28 @@ package eval
 import (
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"strings"
+	"sync"
 
 	"github.com/pkg/errors"
 	"github.com/splunk/qbec/internal/model"
 	"github.com/splunk/qbec/internal/sio"
 	"github.com/splunk/qbec/internal/vm"
 )
 
+const (
+	defaultConcurrency = 5
+	maxDisplayErrors   = 3
+)
+
 // Context is the evaluation context
 type Context struct {
-	App     string // the application for which the evaluation is done
-	Env     string // the environment for which the evaluation is done
-	VM      *vm.VM // the base VM to use for eval
-	Verbose bool   // show generated code
+	App         string // the application for which the evaluation is done
+	Env         string // the environment for which the evaluation is done
+	VM          *vm.VM // the base VM to use for eval
+	Verbose     bool   // show generated code
+	Concurrency int    // concurrent components to evaluate, default 5
 }
 
 // Components evaluates the specified components using the specific runtime
@@ -42,11 +50,11 @@ func Components(components []model.Component, ctx Context) ([]model.K8sLocalObje
 	if ctx.VM == nil {
 		ctx.VM = vm.New(vm.Config{})
 	}
-	cCode, err := evalComponents(components, ctx)
+	componentMap, err := evalComponents(components, ctx)
 	if err != nil {
-		return nil, errors.Wrap(err, "evaluate components")
+		return nil, err
 	}
-	objs, err := k8sObjectsFromJSONString(cCode, ctx.App, ctx.Env)
+	objs, err := k8sObjectsFromJSON(componentMap, ctx.App, ctx.Env)
 	if err != nil {
 		return nil, errors.Wrap(err, "extract objects")
 	}
@@ -80,34 +88,87 @@ func Params(file string, ctx Context) (map[string]interface{}, error) {
 	return ret, nil
 }
 
-func evalComponents(list []model.Component, ctx Context) (string, error) {
-	cfg := ctx.VM.Config().WithVars(map[string]string{model.QbecNames.EnvVarName: ctx.Env})
-	jvm := vm.New(cfg)
-	var lines []string
-	for _, c := range list {
-		switch {
-		case strings.HasSuffix(c.File, ".yaml"):
-			lines = append(lines, fmt.Sprintf("'%s': parseYaml(importstr '%s')", c.Name, c.File))
-		case strings.HasSuffix(c.File, ".json"):
-			lines = append(lines, fmt.Sprintf("'%s': parseJson(importstr '%s')", c.Name, c.File))
-		default:
-			lines = append(lines, fmt.Sprintf("'%s': import '%s'", c.Name, c.File))
+func evalComponent(jvm *vm.VM, c model.Component) (interface{}, error) {
+	var inputCode string
+	contextFile := c.File
+	switch {
+	case strings.HasSuffix(c.File, ".yaml"):
+		inputCode = fmt.Sprintf("std.native('parseYaml')(importstr '%s')", c.File)
+		contextFile = "yaml-loader.jsonnet"
+	case strings.HasSuffix(c.File, ".json"):
+		inputCode = fmt.Sprintf("std.native('parseJson')(importstr '%s')", c.File)
+		contextFile = "json-loader.jsonnet"
+	default:
+		b, err := ioutil.ReadFile(c.File)
+		if err != nil {
+			return nil, errors.Wrap(err, "read inputCode for "+c.File)
 		}
+		inputCode = string(b)
 	}
-	preamble := []string{
-		"local parseYaml = std.native('parseYaml');",
-		"local parseJson = std.native('parseJson');",
+	evalCode, err := jvm.EvaluateSnippet(contextFile, inputCode)
+	if err != nil {
+		return nil, errors.Wrap(err, fmt.Sprintf("evaluate '%s'", c.Name))
 	}
-	code := strings.Join(preamble, "\n") + "\n{\n  " + strings.Join(lines, ",\n  ") + "\n}"
-	if ctx.Verbose {
-		sio.Debugln("Eval components:\n" + code)
+	var data interface{}
+	if err := json.Unmarshal([]byte(evalCode), &data); err != nil {
+		return nil, errors.Wrap(err, fmt.Sprintf("unexpected unmarshal '%s'", c.File))
 	}
-	ret, err := jvm.EvaluateSnippet("component-loader.jsonnet", code)
-	if err != nil {
-		return "", err
+	return data, nil
+}
+
+func evalComponents(list []model.Component, ctx Context) (map[string]interface{}, error) {
+	ret := map[string]interface{}{}
+	if len(list) == 0 {
+		return ret, nil
 	}
-	if ctx.Verbose {
-		sio.Debugln("Eval components output:\n" + prettyJSON(ret))
+
+	ch := make(chan model.Component, len(list))
+	for _, c := range list {
+		ch <- c
+	}
+	close(ch)
+
+	var errs []error
+	var l sync.Mutex
+
+	concurrency := ctx.Concurrency
+	if concurrency <= 0 {
+		concurrency = defaultConcurrency
+	}
+	if concurrency > len(list) {
+		concurrency = len(list)
+	}
+	var wg sync.WaitGroup
+	wg.Add(concurrency)
+
+	cfg := ctx.VM.Config().WithVars(map[string]string{model.QbecNames.EnvVarName: ctx.Env})
+	for i := 0; i < concurrency; i++ {
+		go func() {
+			defer wg.Done()
+			jvm := vm.New(cfg)
+			for c := range ch {
+				obj, err := evalComponent(jvm, c)
+				l.Lock()
+				if err != nil {
+					errs = append(errs, err)
+				} else {
+					ret[c.Name] = obj
+				}
+				l.Unlock()
+			}
+		}()
+	}
+	wg.Wait()
+	if len(errs) > 0 {
+		var msgs []string
+		for i, e := range errs {
+			if i == maxDisplayErrors {
+				msgs = append(msgs, fmt.Sprintf("... and %d more errors", len(errs)-maxDisplayErrors))
+				break
+			}
+			msgs = append(msgs, e.Error())
+		}
+		return nil, errors.New(strings.Join(msgs, "\n"))
 	}
 	return ret, nil
 }

internal/eval/eval_test.go

@@ -88,6 +88,117 @@ func TestEvalComponents(t *testing.T) {
 	a.Equal("jsonnet-config-map", obj.GetName())
 }
 
+func TestEvalComponentsEdges(t *testing.T) {
+	goodComponents := []model.Component{
+		{Name: "g1", File: "testdata/good-components/g1.jsonnet"},
+		{Name: "g2", File: "testdata/good-components/g2.jsonnet"},
+		{Name: "g3", File: "testdata/good-components/g3.jsonnet"},
+		{Name: "g4", File: "testdata/good-components/g4.jsonnet"},
+		{Name: "g5", File: "testdata/good-components/g5.jsonnet"},
+	}
+	goodAssert := func(t *testing.T, ret map[string]interface{}, err error) {
+		require.NotNil(t, err)
+	}
+	tests := []struct {
+		name        string
+		components  []model.Component
+		asserter    func(*testing.T, map[string]interface{}, error)
+		concurrency int
+	}{
+		{
+			name: "no components",
+			asserter: func(t *testing.T, ret map[string]interface{}, err error) {
+				require.Nil(t, err)
+				assert.Equal(t, 0, len(ret))
+			},
+		},
+		{
+			name:       "single bad",
+			components: []model.Component{{Name: "e1", File: "testdata/bad-components/e1.jsonnet"}},
+			asserter: func(t *testing.T, ret map[string]interface{}, err error) {
+				require.NotNil(t, err)
+				assert.Contains(t, err.Error(), "evaluate 'e1'")
+			},
+		},
+		{
+			name: "two bad",
+			components: []model.Component{
+				{Name: "e1", File: "testdata/bad-components/e1.jsonnet"},
+				{Name: "e2", File: "testdata/bad-components/e2.jsonnet"},
+			},
+			asserter: func(t *testing.T, ret map[string]interface{}, err error) {
+				require.NotNil(t, err)
+				assert.Contains(t, err.Error(), "evaluate 'e1'")
+				assert.Contains(t, err.Error(), "evaluate 'e2'")
+			},
+		},
+		{
+			name: "many bad",
+			components: []model.Component{
+				{Name: "e1", File: "testdata/bad-components/e1.jsonnet"},
+				{Name: "e2", File: "testdata/bad-components/e2.jsonnet"},
+				{Name: "e3", File: "testdata/bad-components/e3.jsonnet"},
+				{Name: "e4", File: "testdata/bad-components/e4.jsonnet"},
+				{Name: "e5", File: "testdata/bad-components/e5.jsonnet"},
+			},
+			asserter: func(t *testing.T, ret map[string]interface{}, err error) {
+				require.NotNil(t, err)
+				assert.Contains(t, err.Error(), "... and 2 more errors")
+			},
+		},
+		{
+			name: "bad file",
+			components: []model.Component{
+				{Name: "e1", File: "testdata/bad-components/XXX.jsonnet"},
+			},
+			asserter: func(t *testing.T, ret map[string]interface{}, err error) {
+				require.NotNil(t, err)
+				assert.Contains(t, err.Error(), "no such file")
+			},
+		},
+		{
+			name:        "negative concurrency",
+			components:  goodComponents,
+			asserter:    goodAssert,
+			concurrency: -10,
+		},
+		{
+			name:        "zero concurrency",
+			components:  goodComponents,
+			asserter:    goodAssert,
+			concurrency: 0,
+		},
+		{
+			name:        "4 concurrency",
+			components:  goodComponents,
+			asserter:    goodAssert,
+			concurrency: 4,
+		},
+		{
+			name:        "one concurrency",
+			components:  goodComponents,
+			asserter:    goodAssert,
+			concurrency: 1,
+		},
+		{
+			name:        "million concurrency",
+			components:  goodComponents,
+			asserter:    goodAssert,
+			concurrency: 1000000,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ret, err := evalComponents(test.components, Context{
+				Env:         "dev",
+				VM:          vm.New(vm.Config{}),
+				Concurrency: test.concurrency,
+			})
+			test.asserter(t, ret, err)
+		})
+	}
+}
+
 func TestEvalComponentsBadJson(t *testing.T) {
 	_, err := Components([]model.Component{
 		{

internal/eval/object-extract.go

@@ -22,7 +22,6 @@ import (
 	"reflect"
 	"sort"
 
-	"github.com/pkg/errors"
 	"github.com/splunk/qbec/internal/model"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
@@ -102,11 +101,7 @@ func (w *walker) walkObjects(path string, component string, data interface{}) ([
 	return ret, nil
 }
 
-func k8sObjectsFromJSONString(str string, app, env string) ([]model.K8sLocalObject, error) {
-	var data interface{}
-	if err := json.Unmarshal([]byte(str), &data); err != nil {
-		return nil, errors.Wrap(err, "JSON unmarshal")
-	}
+func k8sObjectsFromJSON(data map[string]interface{}, app, env string) ([]model.K8sLocalObject, error) {
 	w := walker{app: app, env: env, data: data}
 	ret, err := w.walk()
 	if err != nil {

internal/eval/testdata/bad-components/e1.jsonnet

@@ -0,0 +1,5 @@
+{
+  apiVersion: 'v1, //deliberate syntax error
+  kind: 'ConfigMap',
+}
+

internal/eval/testdata/bad-components/e2.jsonnet

@@ -0,0 +1,5 @@
+{
+  apiVersion: 'v1, //deliberate syntax error
+  kind: 'ConfigMap',
+}
+

internal/eval/testdata/bad-components/e3.jsonnet

@@ -0,0 +1,5 @@
+{
+  apiVersion: 'v1, //deliberate syntax error
+  kind: 'ConfigMap',
+}
+

internal/eval/testdata/bad-components/e4.jsonnet

@@ -0,0 +1,5 @@
+{
+  apiVersion: 'v1, //deliberate syntax error
+  kind: 'ConfigMap',
+}
+

internal/eval/testdata/bad-components/e5.jsonnet

@@ -0,0 +1,5 @@
+{
+  apiVersion: 'v1, //deliberate syntax error
+  kind: 'ConfigMap',
+}
+

internal/eval/testdata/good-components/g1.jsonnet

@@ -0,0 +1,12 @@
+{
+  apiVersion: 'v1',
+  kind: 'ConfigMap',
+  metadata: {
+    name: 'g1',
+  }
+  data: {
+    foo: 'bar',
+  },
+}
+
+