Spec validator, better errors, apply waits for previous build (#560)

Adds a validator, invoked through `fission spec validate` and before `fission spec apply`. The validator checks for duplicate names, dangling references, and unused resources.

Also attempts to fix a race between CLI and builder.  Avoids replacing a package while it's building, instead waiting for the build status to change from pending before updating it.

fission/spec.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/fsnotify/fsnotify"
 	"github.com/ghodss/yaml"
+	multierror "github.com/hashicorp/go-multierror"
 	"github.com/mholt/archiver"
 	"github.com/pkg/errors"
 	"github.com/satori/go.uuid"
@@ -100,7 +101,7 @@ type (
 		messageQueueTriggers    []crd.MessageQueueTrigger
 		archiveUploadSpecs      []ArchiveUploadSpec
 
-		sourceMap SourceMap
+		sourceMap sourceMap
 	}
 
 	resourceApplyStatus struct {
@@ -109,8 +110,13 @@ type (
 		deleted []*metav1.ObjectMeta
 	}
 
-	SourceMap struct {
+	location struct {
-		// xxx
+		path string
+		line int
+	}
+	sourceMap struct {
+		// kind -> namespace -> name -> location
+		locations map[string](map[string](map[string]location))
 	}
 )
 
@@ -190,28 +196,205 @@ func specInit(c *cli.Context) error {
 	return nil
 }
 
+// validateFunctionReference checks a function reference
+func (fr *FissionResources) validateFunctionReference(functions map[string]bool, kind string, meta *metav1.ObjectMeta, funcRef fission.FunctionReference) error {
+	if funcRef.Type == fission.FunctionReferenceTypeFunctionName {
+		// triggers only reference functions in their own namespace
+		namespace := meta.Namespace
+		name := funcRef.Name
+		m := &metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		}
+		if _, ok := functions[mapKey(m)]; !ok {
+			return fmt.Errorf("%v: %v '%v' references unknown function '%v'",
+				fr.sourceMap.locations[kind][meta.Namespace][meta.Name],
+				kind,
+				meta.Name,
+				name)
+		} else {
+			functions[mapKey(m)] = true
+		}
+	}
+	return nil
+}
+
 // specValidate parses a set of specs and checks for references to
 // resources that don't exist.
 func specValidate(c *cli.Context) error {
-	//specDir := getSpecDir(c)
+	// this will error on parse errors and on duplicates
+	specDir := getSpecDir(c)
+	fr, err := readSpecs(specDir)
+	checkErr(err, "read specs")
 
-	// parse all specs
+	// this does the rest of the checks, like dangling refs
-	// verify references:
+	err = fr.validate()
-	//   functions from triggers
+	if err != nil {
-	//   packages from functions
+		fmt.Printf("Error validating specs: %v", err)
+	}
 
-	// find unreferenced uploads
+	return nil
+}
+
+func (fr *FissionResources) validate() error {
+	var result *multierror.Error
+
+	// check references: both dangling refs + garbage
+	//   packages -> archives
+	//   functions -> packages
+	//   functions -> environments [TODO]
+	//   triggers -> functions
+
+	// index archives
+	archives := make(map[string]bool)
+	for _, a := range fr.archiveUploadSpecs {
+		archives[a.Name] = false
+	}
+
+	// index packages, check outgoing refs, mark archives that are referenced
+	packages := make(map[string]bool)
+	for _, p := range fr.packages {
+		packages[mapKey(&p.Metadata)] = false
+
+		// check archive refs from package
+		aname := strings.TrimPrefix(p.Spec.Source.URL, ARCHIVE_URL_PREFIX)
+		if len(aname) > 0 {
+			if _, ok := archives[aname]; !ok {
+				result = multierror.Append(result, fmt.Errorf(
+					"%v: package '%v' references unknown source archive %v%v",
+					fr.sourceMap.locations["Package"][p.Metadata.Namespace][p.Metadata.Name],
+					p.Metadata.Name,
+					ARCHIVE_URL_PREFIX,
+					aname))
+			} else {
+				archives[aname] = true
+			}
+		}
+
+		aname = strings.TrimPrefix(p.Spec.Deployment.URL, ARCHIVE_URL_PREFIX)
+		if len(aname) > 0 {
+			if _, ok := archives[aname]; !ok {
+				result = multierror.Append(result, fmt.Errorf(
+					"%v: package '%v' references unknown deployment archive %v%v",
+					fr.sourceMap.locations["Package"][p.Metadata.Namespace][p.Metadata.Name],
+					p.Metadata.Name,
+					ARCHIVE_URL_PREFIX,
+					aname))
+			} else {
+				archives[aname] = true
+			}
+		}
+	}
+
+	// error on unreferenced archives
+	for name, referenced := range archives {
+		if !referenced {
+			result = multierror.Append(result, fmt.Errorf(
+				"%v: archive '%v' is not used in any package",
+				fr.sourceMap.locations["ArchiveUploadSpec"][""][name],
+				name))
+		}
+	}
+
+	// index functions, check function package refs, mark referenced packages
+	functions := make(map[string]bool)
+	for _, f := range fr.functions {
+		functions[mapKey(&f.Metadata)] = false
+
+		// check package ref from function
+		pkgMeta := &metav1.ObjectMeta{
+			Name:      f.Spec.Package.PackageRef.Name,
+			Namespace: f.Spec.Package.PackageRef.Namespace,
+		}
+		if _, ok := packages[mapKey(pkgMeta)]; !ok {
+			result = multierror.Append(result, fmt.Errorf(
+				"%v: function '%v' references unknown package %v/%v",
+				fr.sourceMap.locations["Function"][f.Metadata.Namespace][f.Metadata.Name],
+				f.Metadata.Name,
+				pkgMeta.Namespace,
+				pkgMeta.Name))
+		} else {
+			packages[mapKey(pkgMeta)] = true
+		}
+	}
+
+	// error on unreferenced packages
+	for key, referenced := range packages {
+		ks := strings.Split(key, ":")
+		namespace, name := ks[0], ks[1]
+		if !referenced {
+			result = multierror.Append(result, fmt.Errorf(
+				"%v: package '%v' is not used in any function",
+				fr.sourceMap.locations["Package"][namespace][name],
+				name))
+		}
+	}
+
+	// check function refs from triggers
+	for _, t := range fr.httpTriggers {
+		err := fr.validateFunctionReference(functions, t.Kind, &t.Metadata, t.Spec.FunctionReference)
+		if err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+	for _, t := range fr.kubernetesWatchTriggers {
+		err := fr.validateFunctionReference(functions, t.Kind, &t.Metadata, t.Spec.FunctionReference)
+		if err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+	for _, t := range fr.timeTriggers {
+		err := fr.validateFunctionReference(functions, t.Kind, &t.Metadata, t.Spec.FunctionReference)
+		if err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+	for _, t := range fr.messageQueueTriggers {
+		err := fr.validateFunctionReference(functions, t.Kind, &t.Metadata, t.Spec.FunctionReference)
+		if err != nil {
+			result = multierror.Append(result, err)
+		}
+	}
+
+	// we do not error on unreferenced functions (you can call a function through workflows,
+	// `fission function test`, etc.)
+
+	// (ErrorOrNil returns nil if there were no errors appended.)
+	return result.ErrorOrNil()
+}
+
+func (loc location) String() string {
+	return fmt.Sprintf("%v:%v", loc.path, loc.line)
+}
+
+// Keep track of source location of resources, and track duplicates
+func (fr *FissionResources) trackSourceMap(kind string, newobj *metav1.ObjectMeta, loc *location) error {
+	if _, exists := fr.sourceMap.locations[kind]; !exists {
+		fr.sourceMap.locations[kind] = make(map[string](map[string]location))
+	}
+	if _, exists := fr.sourceMap.locations[kind][newobj.Namespace]; !exists {
+		fr.sourceMap.locations[kind][newobj.Namespace] = make(map[string]location)
+	}
+
+	// check for duplicate resources
+	oldloc, exists := fr.sourceMap.locations[kind][newobj.Namespace][newobj.Name]
+	if exists {
+		return fmt.Errorf("%v: Duplicate %v '%v', first defined in %v", loc, kind, newobj.Name, oldloc)
+	}
+
+	// track new resource
+	fr.sourceMap.locations[kind][newobj.Namespace][newobj.Name] = *loc
 
 	return nil
 }
 
 // parseYaml takes one yaml document, figures out its type, parses it, and puts it in
 // the right list in the given fission resources set.
-func parseYaml(path string, b []byte, fr *FissionResources) error {
+func (fr *FissionResources) parseYaml(b []byte, loc *location) error {
+	var m *metav1.ObjectMeta
 
 	// Figure out the object type by unmarshaling into the TypeMeta struct; then
-	// unmarshal again into the "real" struct once we know the type.  There's almost
+	// unmarshal again into the "real" struct once we know the type.
-	// certainly a better way to do this...
 	var tm TypeMeta
 	err := yaml.Unmarshal(b, &tm)
 	switch tm.Kind {
@@ -219,60 +402,63 @@ func parseYaml(path string, b []byte, fr *FissionResources) error {
 		var v crd.Package
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
+		m = &v.Metadata
 		fr.packages = append(fr.packages, v)
 	case "Function":
 		var v crd.Function
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
+		m = &v.Metadata
 		fr.functions = append(fr.functions, v)
 	case "Environment":
 		var v crd.Environment
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
+		m = &v.Metadata
 		fr.environments = append(fr.environments, v)
 	case "HTTPTrigger":
 		var v crd.HTTPTrigger
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
+
+		// TODO move to validator
 		if !strings.HasPrefix(v.Spec.RelativeURL, "/") {
 			v.Spec.RelativeURL = fmt.Sprintf("/%s", v.Spec.RelativeURL)
 		}
+
+		m = &v.Metadata
 		fr.httpTriggers = append(fr.httpTriggers, v)
 	case "KubernetesWatchTrigger":
 		var v crd.KubernetesWatchTrigger
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
+		m = &v.Metadata
 		fr.kubernetesWatchTriggers = append(fr.kubernetesWatchTriggers, v)
 	case "TimeTrigger":
 		var v crd.TimeTrigger
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
+		m = &v.Metadata
 		fr.timeTriggers = append(fr.timeTriggers, v)
 	case "MessageQueueTrigger":
 		var v crd.MessageQueueTrigger
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
+		m = &v.Metadata
 		fr.messageQueueTriggers = append(fr.messageQueueTriggers, v)
 
 	// The following are not CRDs
@@ -281,22 +467,32 @@ func parseYaml(path string, b []byte, fr *FissionResources) error {
 		var v DeploymentConfig
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
 		}
 		fr.deploymentConfig = v
 	case "ArchiveUploadSpec":
 		var v ArchiveUploadSpec
 		err = yaml.Unmarshal(b, &v)
 		if err != nil {
-			warn(fmt.Sprintf("Failed to parse %v in %v: %v", tm.Kind, path, err))
+			return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc))
-			return err
+		}
+		m = &metav1.ObjectMeta{
+			Name:      v.Name,
+			Namespace: "",
 		}
 		fr.archiveUploadSpecs = append(fr.archiveUploadSpecs, v)
 	default:
 		// no need to error out just because there's some extra files around;
 		// also good for compatibility.
-		warn(fmt.Sprintf("Ignoring unknown type %v in %v", tm.Kind, path))
+		warn(fmt.Sprintf("Ignoring unknown type %v in %v", tm.Kind, loc))
+	}
+
+	// add to source map, check for duplicates
+	if m != nil {
+		err = fr.trackSourceMap(tm.Kind, m, loc)
+		if err != nil {
+			return err
+		}
 	}
 
 	return nil
@@ -313,8 +509,14 @@ func readSpecs(specDir string) (*FissionResources, error) {
 		kubernetesWatchTriggers: make([]crd.KubernetesWatchTrigger, 0),
 		timeTriggers:            make([]crd.TimeTrigger, 0),
 		messageQueueTriggers:    make([]crd.MessageQueueTrigger, 0),
+
+		sourceMap: sourceMap{
+			locations: make(map[string](map[string](map[string]location))),
+		},
 	}
 
+	var result *multierror.Error
+
 	// Users can organize the specdir into subdirs if they want to.
 	err := filepath.Walk(specDir, func(path string, info os.FileInfo, err error) error {
 		// For now just read YAML files. We'll add jsonnet at some point. Skip
@@ -325,26 +527,38 @@ func readSpecs(specDir string) (*FissionResources, error) {
 		// read
 		b, err := ioutil.ReadFile(path)
 		if err != nil {
-			return err
+			result = multierror.Append(result, err)
+			return nil
 		}
 		// handle the case where there are multiple YAML docs per file. go-yaml
 		// doesn't support this directly, yet.
 		docs := bytes.Split(b, []byte("\n---"))
+		lines := 1
 		for _, doc := range docs {
 			d := []byte(strings.TrimSpace(string(doc)))
 			if len(d) != 0 {
 				// parse this document and add whatever is in it to fr
-				err = parseYaml(path, d, &fr)
+				err = fr.parseYaml(d, &location{
+					path: path,
+					line: lines,
+				})
 				if err != nil {
-					return err
+					// collect all errors so user can fix them all
+					result = multierror.Append(result, err)
 				}
 			}
+			// the separator occupies one line, hence the +1
+			lines += strings.Count(string(doc), "\n") + 1
 		}
 		return nil
 	})
+
 	if err != nil {
 		return nil, err
 	}
+	if err := result.ErrorOrNil(); err != nil {
+		return nil, err
+	}
 
 	return &fr, nil
 }
@@ -422,6 +636,10 @@ func specApply(c *cli.Context) error {
 		fr, err := readSpecs(specDir)
 		checkErr(err, "read specs")
 
+		// validate
+		err = fr.validate()
+		checkErr(err, "validate specs")
+
 		// make changes to the cluster based on the specs
 		pkgMetas, as, err := applyResources(fclient, specDir, fr, deleteResources)
 		checkErr(err, "apply specs")
@@ -790,6 +1008,27 @@ func hasDeploymentConfig(m *metav1.ObjectMeta, fr *FissionResources) bool {
 	return false
 }
 
+func waitForPackageBuild(fclient *client.Client, pkg *crd.Package) (*crd.Package, error) {
+	start := time.Now()
+	for {
+		if pkg.Status.BuildStatus != fission.BuildStatusRunning {
+			return pkg, nil
+		}
+		if time.Since(start) > 5*time.Minute {
+			return nil, fmt.Errorf("Package %v has been building for a while.  Giving up on waiting for it.", pkg.Metadata.Name)
+		}
+
+		// TODO watch instead
+		time.Sleep(time.Second)
+
+		var err error
+		pkg, err = fclient.PackageGet(&pkg.Metadata)
+		if err != nil {
+			return nil, err
+		}
+	}
+}
+
 func applyPackages(fclient *client.Client, fr *FissionResources, delete bool) (map[string]metav1.ObjectMeta, *resourceApplyStatus, error) {
 	// get list
 	allObjs, err := fclient.PackageList()
@@ -846,9 +1085,20 @@ func applyPackages(fclient *client.Client, fr *FissionResources, delete bool) (m
 			} else {
 				// update
 				o.Metadata.ResourceVersion = existingObj.Metadata.ResourceVersion
-				newmeta, err := fclient.PackageUpdate(&o)
+
+				// We may be racing against the package builder to update the
+				// package (a previous version might have been getting built).  So,
+				// wait for the package to have a non-running build status.
+				pkg, err := waitForPackageBuild(fclient, &o)
+				if err != nil {
+					// log and ignore
+					fmt.Printf("Error waiting for package '%v' build, ignoring\n", o.Metadata.Name)
+				}
+
+				newmeta, err := fclient.PackageUpdate(pkg)
 				if err != nil {
 					return nil, nil, err
+					// TODO check for resourceVersion conflict errors and retry
 				}
 				ras.updated = append(ras.updated, newmeta)
 				// keep track of metadata in case we need to create a reference to it

glide.lock

@@ -1,5 +1,5 @@
-hash: 3b15246c5a7ca26271ef45e705c92351c786cfbf8ca1c093026b458e527ceefa
+hash: d77d204547318863d07beea96b4e29a90f0d6c106a957dd0ba40eb0345a77e42
-updated: 2018-02-10T19:04:05.240303+01:00
+updated: 2018-03-23T17:49:04.748453691-07:00
 imports:
 - name: cloud.google.com/go
   version: 3b1ae45394a234c385be014e9a488f2bb6eef821
@@ -18,7 +18,7 @@ imports:
   - autorest/azure
   - autorest/date
 - name: github.com/coreos/etcd
-  version: 9c6d93056575da4da94382f473b0ecfcd9c1443b
+  version: 6a265731e10a5137b991c1aa3a83ecefdd149d50
   subpackages:
   - client
 - name: github.com/davecgh/go-spew
@@ -55,7 +55,7 @@ imports:
 - name: github.com/emicklei/go-restful-swagger12
   version: dcef7f55730566d41eae5db10e7d6981829720f6
 - name: github.com/fsnotify/fsnotify
-  version: c2828203cd70a50dcccfb2761f8b1f8ceef9a8e9
+  version: 4da3e2cfbabc9f751898f250b49f2439785783a1
 - name: github.com/ghodss/yaml
   version: 73d445a93680fa1a78ae23a5839bad48f32ba1ee
 - name: github.com/go-openapi/analysis
@@ -97,6 +97,10 @@ imports:
   version: abb68c488872b06c5453865fa59f4818b4ea13a4
   subpackages:
   - local
+- name: github.com/hashicorp/errwrap
+  version: 7554cd9344cec97297fa6649b055a8c98c2a1e55
+- name: github.com/hashicorp/go-multierror
+  version: b7773ae218740a7be65057fc60b366a49b538a44
 - name: github.com/hashicorp/golang-lru
   version: a0d98a5f288019575c6d1f4bb1573fef2d1fcdc4
   subpackages:
@@ -133,7 +137,7 @@ imports:
   subpackages:
   - pb
 - name: github.com/nats-io/nats-streaming-server
-  version: 33414c6f2179201f7fda8743ba89d07184e3fa77
+  version: 7889f37a10062ff8eb9ac11855ed51cddeaf4469
   subpackages:
   - spb
   - util
@@ -148,7 +152,7 @@ imports:
   subpackages:
   - xxHash32
 - name: github.com/pkg/errors
-  version: 30136e27e2ac8d167177e8a583aa4c3fea5be833
+  version: f15c970de5b76fac0b59abb32d62c17cc7bed265
 - name: github.com/PuerkitoBio/purell
   version: 8a290539e2e8629dbc4e6bad948158f790ec31f4
 - name: github.com/PuerkitoBio/urlesc
@@ -237,7 +241,7 @@ imports:
 - name: k8s.io/api
   version: 4b8fc5be9b77d91bbb6525d18591c43699a2b4e5
 - name: k8s.io/apiextensions-apiserver
-  version: fcd622fe88a4a6efcb5aea9e94ee87324ac1b036
+  version: 19d3c0f1ccfb3e4180400ff7c22fe3c879771807
   subpackages:
   - pkg/apis/apiextensions
   - pkg/apis/apiextensions/v1beta1
@@ -245,7 +249,7 @@ imports:
   - pkg/client/clientset/clientset/scheme
   - pkg/client/clientset/clientset/typed/apiextensions/v1beta1
 - name: k8s.io/apimachinery
-  version: 8ab5f3d8a330c2e9baaf84e39042db8d49034ae2
+  version: 208a6980b14bbb263f29482482eabdbcfff9f7bb
   subpackages:
   - pkg/api/equality
   - pkg/api/errors
@@ -278,22 +282,22 @@ imports:
   - pkg/util/diff
   - pkg/util/errors
   - pkg/util/framer
+  - pkg/util/httpstream
+  - pkg/util/httpstream/spdy
   - pkg/util/intstr
   - pkg/util/json
   - pkg/util/net
   - pkg/util/rand
+  - pkg/util/remotecommand
   - pkg/util/runtime
   - pkg/util/sets
   - pkg/util/validation
   - pkg/util/validation/field
   - pkg/util/wait
   - pkg/util/yaml
-  - pkg/util/httpstream
-  - pkg/util/httpstream/spdy
-  - pkg/util/remotecommand
-  - third_party/forked/golang/netutil
   - pkg/version
   - pkg/watch
+  - third_party/forked/golang/netutil
   - third_party/forked/golang/reflect
 - name: k8s.io/client-go
   version: d92e8497f71b7b4e0494e5bd204b48d34bd6f254
@@ -374,11 +378,12 @@ imports:
   - tools/clientcmd/api
   - tools/clientcmd/api/latest
   - tools/clientcmd/api/v1
+  - tools/metrics
   - tools/portforward
   - tools/remotecommand
-  - tools/metrics
   - transport
   - util/cert
+  - util/exec
   - util/flowcontrol
   - util/homedir
   - util/integer

glide.yaml

@@ -60,3 +60,5 @@ import:
   version: ~1.1.0
 - package: github.com/imdario/mergo
   version: ~0.3.2
+- package: github.com/hashicorp/go-multierror
+- package: github.com/hashicorp/errwrap