Improve workflow definition

build/build-env/defaultBuild.sh

@@ -14,7 +14,7 @@ parse() {
 }
 
 echo  "Building with v$(wfcli --version)..."
-if [[ -f $SRC_PKG ]] ; then
+if [[ -f ${SRC_PKG} ]] ; then
     # Package is a single file
     parse ${SRC_PKG} ${DEPLOY_PKG}
 elif [[ -d ${SRC_PKG} ]] ; then
@@ -25,7 +25,7 @@ elif [[ -d ${SRC_PKG} ]] ; then
         parse ${wf} > ${DEPLOY_PKG}/dst
     done
 else
-    echo "Unknown file type: '${SRC_PKG}'"
+    echo "Invalid file type: '${SRC_PKG}'"
     exit 11
 fi
 echo "Build succeeded."

cmd/wfcli/parse.go

@@ -5,22 +5,35 @@ import (
 	"os"
 	"strings"
 
-	"github.com/fission/fission-workflows/pkg/parse/yaml"
+	"github.com/fission/fission-workflows/pkg/parse"
+	"github.com/fission/fission-workflows/pkg/types"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/urfave/cli"
 )
 
 var cmdParse = cli.Command{
-	Name:        "parse",
-	Aliases:     []string{"p"},
-	Usage:       "parse <path-to-yaml> ",
+	Name:    "parse",
+	Aliases: []string{"p"},
+	Usage:   "parse <path-to-source> ",
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "type, t",
+			Value: "yaml",
+			Usage: "Indicate which parser plugin to use for the parsing (yaml|pb).",
+		},
+	},
 	Description: "Read YAML definitions to the executable JSON format (deprecated)",
 	Action: func(c *cli.Context) error {
 
 		if c.NArg() == 0 {
 			panic("Need a path to a yaml workflow definition")
 		}
 
+		parserType := c.String("type")
+		if parserType != "" && !parse.Supports(parserType) {
+			fmt.Printf("Unknown parser '%s'; will try all parsers.", parserType)
+		}
+
 		for _, path := range c.Args() {
 
 			fnName := strings.TrimSpace(path)
@@ -30,21 +43,26 @@ var cmdParse = cli.Command{
 				panic(err)
 			}
 
-			wfSpec, err := yaml.Parse(f)
-			if err != nil {
-				panic(err)
-			}
-
-			marshal := jsonpb.Marshaler{
-				Indent: "  ",
-			}
-			jsonWf, err := marshal.MarshalToString(wfSpec)
+			wfSpec, err := parse.ParseWith(f, parserType)
 			if err != nil {
 				panic(err)
 			}
 
-			fmt.Println(jsonWf)
+			println(toFormattedJson(wfSpec))
 		}
 		return nil
 	},
 }
+
+func toFormattedJson(spec *types.WorkflowSpec) string {
+
+	marshal := jsonpb.Marshaler{
+		Indent: "  ",
+	}
+	jsonWf, err := marshal.MarshalToString(spec)
+	if err != nil {
+		panic(err)
+	}
+
+	return jsonWf
+}

pkg/api/dynamic/api.go

@@ -41,9 +41,8 @@ func (ap *Api) AddDynamicTask(invocationId string, parentId string, taskSpec *ty
 }
 
 func (ap *Api) AddDynamicWorkflow(invocationId string, parentTaskId string, workflowSpec *types.WorkflowSpec) error {
-	taskSpec := types.NewTaskSpec()
 	// TODO add inputs to WorkflowSpec
-	return ap.addDynamicWorkflow(invocationId, parentTaskId, workflowSpec, taskSpec)
+	return ap.addDynamicWorkflow(invocationId, parentTaskId, workflowSpec, &types.TaskSpec{})
 }
 
 func (ap *Api) addDynamicWorkflow(invocationId string, parentTaskId string, wfSpec *types.WorkflowSpec,
@@ -63,9 +62,9 @@ func (ap *Api) addDynamicWorkflow(invocationId string, parentTaskId string, wfSp
 	// Generate Proxy Task
 	proxyTaskSpec := proto.Clone(stubTask).(*types.TaskSpec)
 	proxyTaskSpec.FunctionRef = wfId
-	proxyTaskSpec.AddInput("_parent", typedvalues.ParseString(invocationId))
+	proxyTaskSpec.Input("_parent", typedvalues.ParseString(invocationId))
 	proxyTaskId := parentTaskId + "_child"
-	proxyTask := types.NewTask(proxyTaskId)
+	proxyTask := types.NewTask(proxyTaskId, proxyTaskSpec.FunctionRef)
 	proxyTask.Spec = proxyTaskSpec
 	proxyTask.Status.Status = types.TaskStatus_READY
 	proxyTask.Status.FnRef = workflows.CreateFnRef(wfId)

pkg/apiserver/apiserver.proto

@@ -94,6 +94,13 @@ service WorkflowInvocationAPI {
             get: "/invocation/{id}"
         };
     }
+
+    rpc Validate (WorkflowInvocationSpec) returns (google.protobuf.Empty) {
+        option (google.api.http) = {
+            post: "/invocation/validate"
+            body: "*"
+        };
+    }
 }
 
 message WorkflowInvocationIdentifier {

pkg/apiserver/invocation.go

@@ -2,12 +2,14 @@ package apiserver
 
 import (
 	"errors"
+	"strings"
 	"time"
 
 	"github.com/fission/fission-workflows/pkg/api/invocation"
 	"github.com/fission/fission-workflows/pkg/fes"
 	"github.com/fission/fission-workflows/pkg/types"
 	"github.com/fission/fission-workflows/pkg/types/aggregates"
+	"github.com/fission/fission-workflows/pkg/types/validate"
 	"github.com/golang/protobuf/ptypes/empty"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
@@ -23,6 +25,15 @@ type grpcInvocationApiServer struct {
 	wfiCache fes.CacheReader
 }
 
+func (gi *grpcInvocationApiServer) Validate(ctx context.Context, spec *types.WorkflowInvocationSpec) (*empty.Empty, error) {
+	err := validate.WorkflowInvocationSpec(spec)
+	if err != nil {
+		logrus.Info(strings.Replace(validate.Format(err), "\n", "; ", -1))
+		return nil, err
+	}
+	return &empty.Empty{}, nil
+}
+
 func NewGrpcInvocationApiServer(api *invocation.Api, wfiCache fes.CacheReader) WorkflowInvocationAPIServer {
 	return &grpcInvocationApiServer{api, wfiCache}
 }