diff --git a/fission/spec.go b/fission/spec.go index 633ef8f2..098a050e 100644 --- a/fission/spec.go +++ b/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 { - // xxx + location struct { + 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 - // verify references: - // functions from triggers - // packages from functions + // this does the rest of the checks, like dangling refs + err = fr.validate() + if err != nil { + 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 - // certainly a better way to do this... + // unmarshal again into the "real" struct once we know the type. 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } + 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } + 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } + 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } + + // 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } + 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } + 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } + 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) } 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 err + return errors.Wrap(err, fmt.Sprintf("Failed to parse %v in %v", tm.Kind, loc)) + } + 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 diff --git a/glide.lock b/glide.lock index 7cc6cdfd..bd343307 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 3b15246c5a7ca26271ef45e705c92351c786cfbf8ca1c093026b458e527ceefa -updated: 2018-02-10T19:04:05.240303+01:00 +hash: d77d204547318863d07beea96b4e29a90f0d6c106a957dd0ba40eb0345a77e42 +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 @@ -389,4 +394,4 @@ testImports: subpackages: - difflib - name: github.com/stretchr/objx - version: 8a3f7159479fbc75b30357fbc48f380b7320f08e \ No newline at end of file + version: 8a3f7159479fbc75b30357fbc48f380b7320f08e diff --git a/glide.yaml b/glide.yaml index ba00a7b4..e56abff9 100644 --- a/glide.yaml +++ b/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