Merge pull request #63 from fission/0.1.1

0.1.1

charts/fission-workflows/Chart.yaml

@@ -1,6 +1,7 @@
 apiVersion: v1
 name: fission-workflows
-version: 0.1.0
+version: 0.1.1
+appVersion: 0.1.1
 description: Fission Workflows is a fast workflow engine for serverless functions on Kubernetes
 keywords:
 - fission
@@ -17,4 +18,3 @@ maintainers:
   - name: Soam Vasani
     email: [email protected]
 engine: gotpl
-appVersion: 0.1.0

charts/fission-workflows/requirements.yaml

@@ -1,4 +1,4 @@
 #dependencies:
 #  - name: fission-all
-#    version: ^0.3.0-rc
+#    version: ^0.3.0
 #    repository: https://fission.github.io/fission-charts/

charts/fission-workflows/templates/NOTES.txt

@@ -6,7 +6,7 @@ Usage:
 https://raw.githubusercontent.com/fission/fission-workflows/master/examples/whales/fortune.sh > fortune.sh
 https://raw.githubusercontent.com/fission/fission-workflows/master/examples/whales/whalesay.sh > whalesay.sh
 
-fission env create --name binary --image fission/binary-env:v0.2.1
+fission env create --name binary --image fission/binary-env:v0.3.0
 fission fn create --name whalesay --env binary --code ./whalesay.sh
 fission fn create --name fortune --env binary --code ./fortune.sh
 

charts/fission-workflows/values.yaml

@@ -3,10 +3,10 @@
 #
 
 # Bundle image
-bundleImage: fission/fission-workflows-bundle:0.1.0
+bundleImage: fission/fission-workflows-bundle:0.1.1
 
 # Image of the Fission environment for Fission Workflows
-envImage: fission/workflow-env:0.1.0
+envImage: fission/workflow-env:0.1.1
 
 # Deploy optional Workflow api-server?
 apiserver: true

cmd/fission-workflows-bundle/bundle/bundle.go

@@ -192,12 +192,7 @@ func getWorkflowInvocationCache(ctx context.Context, eventPub pubsub.Publisher)
 }
 
 func setupInternalFunctionRuntime() *native.FunctionEnv {
-	return native.NewFunctionEnv(map[string]native.InternalFunction{
-		"if":      &builtin.FunctionIf{},
-		"noop":    &builtin.FunctionNoop{},
-		"compose": &builtin.FunctionCompose{},
-		"sleep":   &builtin.FunctionSleep{},
-	})
+	return native.NewFunctionEnv(builtin.DefaultBuiltinFunctions)
 }
 
 func setupFissionFunctionRuntime(poolmgrAddr string) *fission.FunctionEnv {

cmd/wfcli/README.md

@@ -0,0 +1,26 @@
+# Fission Workflows CLI (wfcli)
+
+This provides a CLI for Fission Workflows, providing additional functionality for workflows over the regular functions 
+available in the Fission CLI.
+
+_Note: this is an experimental CLI -- in the near future wfcli will be merged into the Fission CLI._
+
+## Installation
+```bash
+go install
+``` 
+
+## Usage
+```bash
+wfcli status # View whether the Fission Workflows deployment can be reached.
+
+wfcli workflow get # List all workflows in the workflow engine.
+
+wfcli workflow get <id> # Get the definition of a specifc workflow
+
+wfcli invocation get # List all invocations so-far (both in-progress and finished)
+
+wfcli invocation get <id> # Get all info of a specific invocation
+
+wfcli invocation status <id> # Get a concise overview of the progress of an invocation 
+```

cmd/wfcli/main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/go-openapi/strfmt"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/urfave/cli"
+	"sort"
 )
 
 // This is a prototype of the CLI (and will be integrated into the Fission CLI eventually).
@@ -69,48 +70,51 @@ func main() {
 			Description: "Parse YAML definitions to the executable JSON format",
 			Action: func(c *cli.Context) error {
 
-				if len(os.Args) == 0 {
+				if c.NArg() == 0 {
 					panic("Need a path to a .yaml")
 				}
 
-				fnName := strings.TrimSpace(os.Args[1])
-				fmt.Printf("Formatting: '%s'\n", fnName)
+				for _, path := range c.Args() {
 
-				if !strings.HasSuffix(fnName, "yaml") {
-					panic("Only YAML workflow definitions are supported")
-				}
+					fnName := strings.TrimSpace(path)
+					fmt.Printf("Formatting: '%s'\n", fnName)
 
-				f, err := os.Open(fnName)
-				if err != nil {
-					panic(err)
-				}
+					if !strings.HasSuffix(fnName, "yaml") {
+						panic("Only YAML workflow definitions are supported")
+					}
 
-				wfDef, err := yaml.Parse(f)
-				if err != nil {
-					panic(err)
-				}
+					f, err := os.Open(fnName)
+					if err != nil {
+						panic(err)
+					}
 
-				wfSpec, err := yaml.Transform(wfDef)
-				if err != nil {
-					panic(err)
-				}
+					wfDef, err := yaml.Parse(f)
+					if err != nil {
+						panic(err)
+					}
 
-				marshal := jsonpb.Marshaler{
-					Indent: "  ",
-				}
-				jsonWf, err := marshal.MarshalToString(wfSpec)
-				if err != nil {
-					panic(err)
-				}
+					wfSpec, err := yaml.Transform(wfDef)
+					if err != nil {
+						panic(err)
+					}
 
-				outputFile := strings.Replace(fnName, "yaml", "json", -1)
+					marshal := jsonpb.Marshaler{
+						Indent: "  ",
+					}
+					jsonWf, err := marshal.MarshalToString(wfSpec)
+					if err != nil {
+						panic(err)
+					}
 
-				err = ioutil.WriteFile(outputFile, []byte(jsonWf), 0644)
-				if err != nil {
-					panic(err)
-				}
+					outputFile := strings.Replace(fnName, "yaml", "json", -1)
+
+					err = ioutil.WriteFile(outputFile, []byte(jsonWf), 0644)
+					if err != nil {
+						panic(err)
+					}
 
-				println(outputFile)
+					println(outputFile)
+				}
 				return nil
 			},
 		},
@@ -133,8 +137,10 @@ func main() {
 							if err != nil {
 								panic(err)
 							}
+							wfs := resp.Payload.Workflows
+							sort.Strings(wfs)
 							rows := [][]string{}
-							for _, wfId := range resp.Payload.Workflows {
+							for _, wfId := range wfs {
 								resp, err := wfApi.Get0(workflow_api.NewGet0Params().WithID(wfId))
 								if err != nil {
 									panic(err)
@@ -208,6 +214,7 @@ func main() {
 								panic(err)
 							}
 							wis := resp.Payload
+							sort.Strings(wis.Invocations)
 							rows := [][]string{}
 							for _, wfiId := range wis.Invocations {
 								resp, err := wfiApi.Get(workflow_invocation_api.NewGetParams().WithID(wfiId))
@@ -312,8 +319,14 @@ func main() {
 	app.Run(os.Args)
 }
 func collectStatus(tasks map[string]models.Task, taskStatus map[string]models.TaskInvocation, rows [][]string) [][]string {
+	ids := []string{}
 	for id := range tasks {
-		status := types.TaskInvocationStatus_UNKNOWN.String()
+		ids = append(ids, id)
+	}
+	sort.Strings(ids)
+
+	for _, id := range ids {
+		status := types.TaskInvocationStatus_SCHEDULED.String()
 		updated := ""
 		started := ""
 

examples/misc/sleepalot.wf.json

@@ -0,0 +1,134 @@
+{
+  "tasks": {
+    "A": {
+      "id": "A",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjEwcyI="
+        }
+      },
+      "requires": {
+      }
+    },
+    "B": {
+      "id": "B",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjVzIg=="
+        }
+      },
+      "requires": {
+        "A": {
+
+        }
+      },
+      "await": 1
+    },
+    "C": {
+      "id": "C",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjVzIg=="
+        }
+      },
+      "requires": {
+        "B": {
+
+        }
+      },
+      "await": 1
+    },
+    "D1": {
+      "id": "D1",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjVzIg=="
+        }
+      },
+      "requires": {
+        "C": {
+
+        }
+      },
+      "await": 1
+    },
+    "D2": {
+      "id": "D2",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjEwcyI="
+        }
+      },
+      "requires": {
+        "C": {
+
+        }
+      },
+      "await": 1
+    },
+    "D3": {
+      "id": "D3",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjIwcyI="
+        }
+      },
+      "requires": {
+        "C": {
+
+        }
+      },
+      "await": 1
+    },
+    "E": {
+      "id": "E",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjVzIg=="
+        }
+      },
+      "requires": {
+        "D1": {
+
+        },
+        "D2": {
+
+        },
+        "D3": {
+
+        }
+      },
+      "await": 3
+    },
+    "F": {
+      "id": "F",
+      "functionRef": "sleep",
+      "inputs": {
+        "default": {
+          "type": "json/string",
+          "value": "IjVzIg=="
+        }
+      },
+      "requires": {
+        "E": {
+
+        }
+      },
+      "await": 1
+    }
+  }
+}

examples/misc/sleepalot.wf.yaml

@@ -0,0 +1,46 @@
+#
+# The purpose of this workflow is to show task dependencies in action
+#
+apiVersion: 1
+# No output expected, so no need to add 'output'
+tasks:
+  A:
+    run: sleep
+    inputs: 10s
+  B:
+    run: sleep
+    inputs: 5s
+    requires:
+    - A
+  C:
+    run: sleep
+    inputs: 5s
+    requires:
+    - B
+  D1:
+    run: sleep
+    inputs: 5s
+    requires:
+    - C
+  D2:
+    run: sleep
+    inputs: 10s
+    requires:
+    - C
+  D3:
+    run: sleep
+    inputs: 20s
+    requires:
+    - C
+  E:
+    run: sleep
+    inputs: 5s
+    requires:
+    - D1
+    - D2
+    - D3
+  F:
+    run: sleep
+    inputs: 5s
+    requires:
+    - E