workflow: remove promise in workflow

masco · masco · commit bf16528a027c · 2019-05-14T15:36:12.000+05:30
diff --git a/alert/server.go b/alert/server.go
@@ -402,14 +402,11 @@ func (a *Server) Stop() {
 func NewServer(apiServer *api.Server, pool ws.StructSpeakerPool, graph *graph.Graph, parser *traversal.GremlinTraversalParser, etcdClient *etcd.Client) (*Server, error) {
 	election := etcdClient.NewElection("alert-server")
 
-	runtime, err := js.NewRuntime()
+	runtime, err := api.NewWorkflowRuntime(graph, parser, apiServer)
 	if err != nil {
 		return nil, err
 	}
 
-	runtime.Start()
-	api.RegisterAPIServer(runtime, graph, parser, apiServer)
-
 	as := &Server{
 		MasterElection: election,
 		Pool:           pool,
diff --git a/api/server/workflow_call.go b/api/server/workflow_call.go
@@ -21,8 +21,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"strings"
 
+	"github.com/gorilla/mux"
 	"github.com/skydive-project/skydive/js"
 	"github.com/skydive-project/skydive/rbac"
 
@@ -38,6 +38,7 @@ type WorkflowCallAPIHandler struct {
 	apiServer *Server
 	graph     *graph.Graph
 	parser    *traversal.GremlinTraversalParser
+	runtime   *js.Runtime
 }
 
 func (wc *WorkflowCallAPIHandler) executeWorkflow(w http.ResponseWriter, r *auth.AuthenticatedRequest) {
@@ -46,29 +47,22 @@ func (wc *WorkflowCallAPIHandler) executeWorkflow(w http.ResponseWriter, r *auth
 		return
 	}
 
-	uriSegments := strings.Split(r.URL.Path, "/")
 	decoder := json.NewDecoder(r.Body)
 	var wfCall types.WorkflowCall
 	if err := decoder.Decode(&wfCall); err != nil {
 		writeError(w, http.StatusBadRequest, err)
 		return
 	}
 
-	workflow, err := wc.getWorkflow(uriSegments[3])
-	if err != nil {
-		writeError(w, http.StatusBadRequest, err)
-		return
-	}
+	vars := mux.Vars(&r.Request)
 
-	runtime, err := js.NewRuntime()
+	workflow, err := wc.getWorkflow(vars["ID"])
 	if err != nil {
-		writeError(w, http.StatusFailedDependency, err)
+		writeError(w, http.StatusBadRequest, err)
 		return
 	}
 
-	runtime.Start()
-	RegisterAPIServer(runtime, wc.graph, wc.parser, wc.apiServer)
-	ottoResult, err := runtime.ExecPromise(workflow.Source, wfCall.Params...)
+	ottoResult, err := wc.runtime.ExecFunction(workflow.Source, wfCall.Params...)
 	if err != nil {
 		writeError(w, http.StatusBadRequest, err)
 		return
@@ -100,7 +94,7 @@ func (wc *WorkflowCallAPIHandler) registerEndPoints(s *shttp.Server, authBackend
 		{
 			Name:        "WorkflowCall",
 			Method:      "POST",
-			Path:        "/api/workflow/{workflowID}/call",
+			Path:        "/api/workflow/{ID}/call",
 			HandlerFunc: wc.executeWorkflow,
 		},
 	}
@@ -109,11 +103,19 @@ func (wc *WorkflowCallAPIHandler) registerEndPoints(s *shttp.Server, authBackend
 }
 
 // RegisterWorkflowCallAPI registers a new workflow  call api handler
-func RegisterWorkflowCallAPI(s *shttp.Server, authBackend shttp.AuthenticationBackend, apiServer *Server, g *graph.Graph, tr *traversal.GremlinTraversalParser) {
+func RegisterWorkflowCallAPI(s *shttp.Server, authBackend shttp.AuthenticationBackend, apiServer *Server, g *graph.Graph, tr *traversal.GremlinTraversalParser) error {
+	runtime, err := NewWorkflowRuntime(g, tr, apiServer)
+	if err != nil {
+		return err
+	}
+
 	workflowCallAPIHandler := &WorkflowCallAPIHandler{
 		apiServer: apiServer,
 		graph:     g,
 		parser:    tr,
+		runtime:   runtime,
 	}
 	workflowCallAPIHandler.registerEndPoints(s, authBackend)
+
+	return nil
 }
diff --git a/api/server/workflow_runtime.go b/api/server/workflow_runtime.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/skydive-project/skydive/js"
 
@@ -30,42 +31,56 @@ import (
 	"github.com/skydive-project/skydive/graffiti/graph/traversal"
 )
 
-// RegisterAPIServer exports Go functions required by the API
-// to run inside the JS VM
-func RegisterAPIServer(r *js.Runtime, g *graph.Graph, gremlinParser *traversal.GremlinTraversalParser, server *Server) {
+// NewWorkflowRuntime returns a new Workflow runtime
+func NewWorkflowRuntime(g *graph.Graph, tr *traversal.GremlinTraversalParser, server *Server) (*js.Runtime, error) {
+	runtime, err := js.NewRuntime()
+	if err != nil {
+		return nil, err
+	}
+	runtime.Start()
+
 	queryGremlin := func(query string) otto.Value {
-		ts, err := gremlinParser.Parse(strings.NewReader(query))
+		ts, err := tr.Parse(strings.NewReader(query))
 		if err != nil {
-			return r.MakeCustomError("ParseError", err.Error())
+			return runtime.MakeCustomError("ParseError", err.Error())
 		}
 
 		result, err := ts.Exec(g, false)
 		if err != nil {
-			return r.MakeCustomError("ExecuteError", err.Error())
+			return runtime.MakeCustomError("ExecuteError", err.Error())
 		}
 
 		source, err := result.MarshalJSON()
 		if err != nil {
-			return r.MakeCustomError("MarshalError", err.Error())
+			return runtime.MakeCustomError("MarshalError", err.Error())
 		}
 
-		r, _ := r.ToValue(string(source))
+		r, _ := runtime.ToValue(string(source))
 		return r
 	}
 
-	r.Set("Gremlin", func(call otto.FunctionCall) otto.Value {
+	runtime.Set("sleep", func(call otto.FunctionCall) otto.Value {
+		if len(call.ArgumentList) != 1 || !call.Argument(0).IsNumber() {
+			return runtime.MakeCustomError("MissingArgument", "Sleep requires a number parameter")
+		}
+		t, _ := call.Argument(0).ToInteger()
+		time.Sleep(time.Duration(t) * time.Millisecond)
+		return otto.NullValue()
+	})
+
+	runtime.Set("Gremlin", func(call otto.FunctionCall) otto.Value {
 		if len(call.ArgumentList) < 1 || !call.Argument(0).IsString() {
-			return r.MakeCustomError("MissingQueryArgument", "Gremlin requires a string parameter")
+			return runtime.MakeCustomError("MissingQueryArgument", "Gremlin requires a string parameter")
 		}
 
 		query := call.Argument(0).String()
 
 		return queryGremlin(query)
 	})
 
-	r.Set("request", func(call otto.FunctionCall) otto.Value {
+	runtime.Set("request", func(call otto.FunctionCall) otto.Value {
 		if len(call.ArgumentList) < 3 || !call.Argument(0).IsString() || !call.Argument(1).IsString() || !call.Argument(2).IsString() {
-			return r.MakeCustomError("WrongArguments", "Import requires 3 string parameters")
+			return runtime.MakeCustomError("WrongArguments", "Import requires 3 string parameters")
 		}
 
 		url := call.Argument(0).String()
@@ -74,15 +89,15 @@ func RegisterAPIServer(r *js.Runtime, g *graph.Graph, gremlinParser *traversal.G
 
 		subs := strings.Split(url, "/") // filepath.Base(url)
 		if len(subs) < 3 {
-			return r.MakeCustomError("WrongArgument", fmt.Sprintf("Malformed URL %s", url))
+			return runtime.MakeCustomError("WrongArgument", fmt.Sprintf("Malformed URL %s", url))
 		}
 		resource := subs[2]
 
 		// For topology query, we directly call the Gremlin engine
 		if resource == "topology" {
 			query := types.TopologyParam{}
 			if err := json.Unmarshal(data, &query); err != nil {
-				return r.MakeCustomError("WrongArgument", fmt.Sprintf("Invalid query %s", string(data)))
+				return runtime.MakeCustomError("WrongArgument", fmt.Sprintf("Invalid query %s", string(data)))
 			}
 
 			return queryGremlin(query.GremlinQuery)
@@ -98,17 +113,17 @@ func RegisterAPIServer(r *js.Runtime, g *graph.Graph, gremlinParser *traversal.G
 		case "POST":
 			res := handler.New()
 			if err := json.Unmarshal([]byte(data), res); err != nil {
-				return r.MakeCustomError("UnmarshalError", err.Error())
+				return runtime.MakeCustomError("UnmarshalError", err.Error())
 			}
 			if err := handler.Create(res); err != nil {
-				return r.MakeCustomError("CreateError", err.Error())
+				return runtime.MakeCustomError("CreateError", err.Error())
 			}
 			b, _ := json.Marshal(res)
 			content = string(b)
 
 		case "DELETE":
 			if len(subs) < 4 {
-				return r.MakeCustomError("WrongArgument", "No ID specified")
+				return runtime.MakeCustomError("WrongArgument", "No ID specified")
 			}
 			handler.Delete(subs[3])
 
@@ -121,7 +136,7 @@ func RegisterAPIServer(r *js.Runtime, g *graph.Graph, gremlinParser *traversal.G
 				id := subs[3]
 				obj, found := handler.Get(id)
 				if !found {
-					return r.MakeCustomError("NotFound", fmt.Sprintf("%s %s could not be found", resource, id))
+					return runtime.MakeCustomError("NotFound", fmt.Sprintf("%s %s could not be found", resource, id))
 				}
 				b, _ := json.Marshal(obj)
 				content = string(b)
@@ -130,9 +145,11 @@ func RegisterAPIServer(r *js.Runtime, g *graph.Graph, gremlinParser *traversal.G
 
 		value, err := otto.ToValue(content)
 		if err != nil {
-			return r.MakeCustomError("WrongValue", err.Error())
+			return runtime.MakeCustomError("WrongValue", err.Error())
 		}
 
 		return value
 	})
+
+	return runtime, nil
 }
diff --git a/js/api.ts b/js/api.ts
@@ -70,22 +70,14 @@ if (jsEnv) {
     // exported by Skydive that makes use of the REST client
     // so we get support for authentication
     makeRequest = function (client: Client, url: string, method: string, body: string, opts: Object) : any {
-        return new Promise(function (resolve, reject) {
-            var err: any
-            var data: any
-            try {
-                var output = request(url, method, body)
-                if (output && output.length > 0) {
-                    data = JSON.parse(output)
-                } else {
-                    data = null
-                }
-                resolve(data)
-            }
-            catch(e) {
-                reject(e)
-            }
-        })
+        var data: any
+        var output = request(url, method, body)
+        if (output && output.length > 0) {
+            data = JSON.parse(output)
+        } else {
+            data = null
+        }
+        return {'then': function() { return data; } }
     }
 }
 
@@ -215,14 +207,10 @@ export class API<T extends APIObject> {
     get(id) {
         let resource = this.Resource;
         let factory = this.Factory;
-
         return this.client.request('/api/' + resource + "/" + id, "GET", "", {})
             .then(function (data) {
                 return SerializationHelper.toInstance(new factory(), data)
             })
-            .catch(function (error) {
-                return Error("Capture not found")
-            })
     }
 
     delete(id: string) {
@@ -325,12 +313,9 @@ export class Step implements Step {
         return this.previous.serialize(data);
     }
 
-    then(cb) {
-        let self = this;
-        return self.api.query(self.toString())
-            .then(function (data) {
-                return cb(self.serialize(data))
-            });
+    result() {
+        let data = this.api.query(this.toString())
+        return this.serialize(data)
     }
 }
 
@@ -937,7 +922,3 @@ export class Client {
         return makeRequest(this, url, method, data, opts);
     }
 }
-
-export function sleep(time) {
-    return new Promise(function(resolve) { return setTimeout(resolve, time); })
-}
diff --git a/js/browser.ts b/js/browser.ts
@@ -37,4 +37,3 @@ window.EdgeRule = apiLib.EdgeRule
 window.NodeRule = apiLib.NodeRule
 window.PacketInjection = apiLib.PacketInjection
 window.Workflow = apiLib.Workflow
-window.sleep = apiLib.sleep
diff --git a/statics/workflows/check-connectivity.yaml b/statics/workflows/check-connectivity.yaml
@@ -11,6 +11,7 @@ Parameters:
     Type: node
 Source: |
     function CheckConnectivity(from, to) {
+      try {
         var capture = new Capture();
         capture.GremlinQuery = "G.V().Has('TID', '" + from + "').ShortestPathTo(Metadata('TID', '" + to + "'))";
 
@@ -21,39 +22,19 @@ Source: |
         packetInjection.ICMPID = Math.floor(Math.random() * 1000);
         packetInjection.Count = 5
 
-        return client.captures.create(capture).then(function (c) {
-            capture = c
-            return sleep(2000)
-        }).then(function () {
-            return client.G.V().Has("TID", from)
-        }).then(function (nodes) {
-            if (nodes[0].Metadata.Neutron && nodes[0].Metadata.Neutron.IPV4) {
-                packetInjection.SrcIP = nodes[0].Metadata.Neutron.IPV4[0]
-            }
-            if (nodes[0].Metadata.ExtID && nodes[0].Metadata.ExtID["attached-mac"]) {
-                packetInjection.SrcMAC = nodes[0].Metadata.ExtID["attached-mac"]
-            }
-            return client.G.V().Has("TID", to)
-        }).then(function (nodes) {
-            if (nodes[0].Metadata.Neutron && nodes[0].Metadata.Neutron.IPV4) {
-                packetInjection.DstIP = nodes[0].Metadata.Neutron.IPV4[0]
-            }
-            if (nodes[0].Metadata.ExtID && nodes[0].Metadata.ExtID["attached-mac"]) {
-                packetInjection.DstMAC = nodes[0].Metadata.ExtID["attached-mac"]
-            }
-            return client.packetInjections.create(packetInjection)
-        }).then(function () {
-            return sleep(2000)
-        }).then(function () {
-            return client.G.Flows().Has("ICMP.ID", packetInjection.ICMPID)
-        }).then(function (flows) {
-            console.log("Flows requested ! :-)")
-            console.log(flows)
-            return {
+        capture = client.captures.create(capture)
+        sleep(1000)
+        client.packetInjections.create(packetInjection)
+        sleep(1000)
+
+        var flows = client.G.Flows().Has("ICMP.ID", packetInjection.ICMPID).result()
+        return {
                 "State": flows.length > 0 && flows[0].Metric.ABPackets > 0 && flows[0].Metric.BAPackets > 0,
                 "Flows": flows
-            }
-        }).finally(function () {
-            client.captures.delete(capture.UUID)
-        })
+        };
+      } catch (e) {
+        console.log(e)
+      } finally {
+        if (capture && capture.UUID) client.captures.delete(capture.UUID)
+      }
     }
diff --git a/statics/workflows/check-mtu.yaml b/statics/workflows/check-mtu.yaml
diff --git a/statics/workflows/flow-matrix.yaml b/statics/workflows/flow-matrix.yaml
diff --git a/tests/workflow_test.go b/tests/workflow_test.go
