Merge pull request #106 from s8sg/execute-operation

Add Execute method for operation with options

faas_operation.go

@@ -1,7 +1,13 @@
 package faasflow
 
 import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
 	"net/http"
+	"net/url"
+	"os"
+	"path"
 	"strings"
 )
 
@@ -107,6 +113,209 @@ func (operation *FaasOperation) GetId() string {
 	return id
 }
 
+func (operation *FaasOperation) Encode() []byte {
+	return []byte("")
+}
+
+// buildURL builds OpenFaaS function execution url for the flow
+func buildURL(gateway, rPath, function string) string {
+	u, _ := url.Parse(gateway)
+	u.Path = path.Join(u.Path, rPath+"/"+function)
+	return u.String()
+}
+
+// makeQueryStringFromParam create query string from provided query
+func makeQueryStringFromParam(params map[string][]string) string {
+	if params == nil {
+		return ""
+	}
+	result := ""
+	for key, array := range params {
+		for _, value := range array {
+			keyVal := fmt.Sprintf("%s-%s", key, value)
+			if result == "" {
+				result = "?" + keyVal
+			} else {
+				result = result + "&" + keyVal
+			}
+		}
+	}
+	return result
+}
+
+// buildHttpRequest build upstream request for function
+func buildHttpRequest(url string, method string, data []byte, params map[string][]string,
+	headers map[string]string) (*http.Request, error) {
+
+	queryString := makeQueryStringFromParam(params)
+	if queryString != "" {
+		url = url + queryString
+	}
+
+	httpReq, err := http.NewRequest(method, url, bytes.NewReader(data))
+	if err != nil {
+		return nil, err
+	}
+
+	for key, value := range headers {
+		httpReq.Header.Add(key, value)
+	}
+
+	return httpReq, nil
+}
+
+// executeFunction executes a function call
+func executeFunction(gateway string, operation *FaasOperation, data []byte) ([]byte, error) {
+	var err error
+	var result []byte
+
+	name := operation.Function
+	params := operation.GetParams()
+	headers := operation.GetHeaders()
+
+	funcUrl := buildURL("http://"+gateway, "function", name)
+
+	method := os.Getenv("default-method")
+	if method == "" {
+		method = "POST"
+	}
+
+	if m, ok := headers["method"]; ok {
+		method = m
+	}
+
+	httpReq, err := buildHttpRequest(funcUrl, method, data, params, headers)
+	if err != nil {
+		return []byte{}, fmt.Errorf("cannot connect to Function on URL: %s", funcUrl)
+	}
+
+	if operation.Requesthandler != nil {
+		operation.Requesthandler(httpReq)
+	}
+
+	client := &http.Client{}
+	resp, err := client.Do(httpReq)
+	if err != nil {
+		return []byte{}, err
+	}
+
+	defer resp.Body.Close()
+	if operation.OnResphandler != nil {
+		result, err = operation.OnResphandler(resp)
+	} else {
+		if resp.StatusCode < 200 || resp.StatusCode > 299 {
+			err = fmt.Errorf("invalid return status %d while connecting %s", resp.StatusCode, funcUrl)
+			result, _ = ioutil.ReadAll(resp.Body)
+		} else {
+			result, err = ioutil.ReadAll(resp.Body)
+		}
+	}
+
+	return result, err
+}
+
+// executeCallback executes a callback
+func executeCallback(operation *FaasOperation, data []byte) error {
+	var err error
+
+	cbUrl := operation.CallbackUrl
+	params := operation.GetParams()
+	headers := operation.GetHeaders()
+
+	method := os.Getenv("default-method")
+	if method == "" {
+		method = "POST"
+	}
+
+	if m, ok := headers["method"]; ok {
+		method = m
+	}
+
+	httpReq, err := buildHttpRequest(cbUrl, method, data, params, headers)
+	if err != nil {
+		return fmt.Errorf("cannot connect to Function on URL: %s", cbUrl)
+	}
+
+	if operation.Requesthandler != nil {
+		operation.Requesthandler(httpReq)
+	}
+
+	client := &http.Client{}
+	resp, err := client.Do(httpReq)
+	if err != nil {
+		return err
+	}
+
+	defer resp.Body.Close()
+	if operation.OnResphandler != nil {
+		_, err = operation.OnResphandler(resp)
+	} else {
+		if resp.StatusCode < 200 || resp.StatusCode > 299 {
+			cbResult, _ := ioutil.ReadAll(resp.Body)
+			err := fmt.Errorf("%v:%s", err, string(cbResult))
+			return err
+		}
+	}
+	return err
+
+}
+
+func (operation *FaasOperation) Execute(data []byte, option map[string]interface{}) ([]byte, error) {
+	var result []byte
+	var err error
+
+	reqId := fmt.Sprintf("%v", option["request-id"])
+	gateway := fmt.Sprintf("%v", option["gateway"])
+
+	switch {
+	// If function
+	case operation.Function != "":
+		fmt.Printf("[Request `%s`] Executing function `%s`\n",
+			reqId, operation.Function)
+		result, err = executeFunction(gateway, operation, data)
+		if err != nil {
+			err = fmt.Errorf("Function(%s), error: function execution failed, %v",
+				operation.Function, err)
+			if operation.FailureHandler != nil {
+				err = operation.FailureHandler(err)
+			}
+			if err != nil {
+				return nil, err
+			}
+		}
+
+	// If callback
+	case operation.CallbackUrl != "":
+		fmt.Printf("[Request `%s`] Executing callback `%s`\n",
+			reqId, operation.CallbackUrl)
+		err = executeCallback(operation, data)
+		if err != nil {
+			err = fmt.Errorf("Callback(%s), error: callback failed, %v",
+				operation.CallbackUrl, err)
+			if operation.FailureHandler != nil {
+				err = operation.FailureHandler(err)
+			}
+			if err != nil {
+				return nil, err
+			}
+		}
+
+	// If modifier
+	default:
+		fmt.Printf("[Request `%s`] Executing modifier\n", reqId)
+		result, err = operation.Mod(data)
+		if err != nil {
+			err = fmt.Errorf("error: Failed at modifier, %v", err)
+			return nil, err
+		}
+		if result == nil {
+			result = []byte("")
+		}
+	}
+
+	return result, nil
+}
+
 func (operation *FaasOperation) GetProperties() map[string][]string {
 
 	result := make(map[string][]string)

sdk/executor/executor.go

@@ -41,9 +41,9 @@ func (req *PartialState) Encode() ([]byte, error) {
 // ExecutionRuntime implements how operation executed and handle next nodes in async
 type ExecutionRuntime interface {
 	// HandleNextNode handles execution of next nodes based on partial state
-	HandleNextNode(*PartialState) (err error)
-	// ExecuteOperation implements execution of an operation
-	ExecuteOperation(sdk.Operation, []byte) ([]byte, error)
+	HandleNextNode(state *PartialState) (err error)
+	// Provide an execution option that will be passed to the operation
+	GetExecutionOption(operation sdk.Operation) map[string]interface{}
 }
 
 // Executor implements a faas-flow executor
@@ -345,10 +345,13 @@ func (fexec *FlowExecutor) executeNode(request []byte) ([]byte, error) {
 		if fexec.executor.MonitoringEnabled() {
 			fexec.eventHandler.ReportOperationStart(operation.GetId(), currentNode.GetUniqueId(), fexec.id)
 		}
+
+		options := fexec.executor.GetExecutionOption(operation)
+
 		if result == nil {
-			result, err = fexec.executor.ExecuteOperation(operation, request)
+			result, err = operation.Execute(request, options)
 		} else {
-			result, err = fexec.executor.ExecuteOperation(operation, result)
+			result, err = operation.Execute(result, options)
 		}
 		if err != nil {
 			if fexec.executor.MonitoringEnabled() {

sdk/operation.go

@@ -4,7 +4,10 @@ import ()
 
 type Operation interface {
 	GetId() string
+	Encode() []byte
 	GetProperties() map[string][]string
+	// Execute executes an operation, executor can pass configuration
+	Execute([]byte, map[string]interface{}) ([]byte, error)
 }
 
 type BlankOperation struct {
@@ -14,6 +17,14 @@ func (ops *BlankOperation) GetId() string {
 	return "end"
 }
 
+func (ops *BlankOperation) Encode() []byte {
+	return []byte("")
+}
+
 func (ops *BlankOperation) GetProperties() map[string][]string {
 	return make(map[string][]string)
 }
+
+func (ops *BlankOperation) Execute(data []byte, option map[string]interface{}) ([]byte, error) {
+	return data, nil
+}

template/faas-flow/handler.go

@@ -31,7 +31,7 @@ var (
 // implements faasflow.EventHandler
 type openFaasEventHandler struct {
 	currentNodeId string        // used to inject current node id in tracer
-	tracer        *traceHandler // handle traces with opentracing
+	tracer        *traceHandler // handle traces with open-tracing
 	flowName      string
 	header        http.Header
 }
@@ -129,63 +129,6 @@ func (eh *openFaasEventHandler) Flush() {
 
 // ExecutionRuntime
 
-func (of *openFaasExecutor) ExecuteOperation(operation sdk.Operation, data []byte) ([]byte, error) {
-	var result []byte
-	var err error
-
-	ofoperation, ok := operation.(*faasflow.FaasOperation)
-	if !ok {
-		return nil, fmt.Errorf("Operation is not of type faasflow.OpenfaasOperation")
-	}
-
-	switch {
-	// If function
-	case ofoperation.Function != "":
-		fmt.Printf("[Request `%s`] Executing function `%s`\n",
-			of.reqId, ofoperation.Function)
-		result, err = executeFunction(of.gateway, ofoperation, data)
-		if err != nil {
-			err = fmt.Errorf("Function(%s), error: function execution failed, %v",
-				ofoperation.Function, err)
-			if ofoperation.FailureHandler != nil {
-				err = ofoperation.FailureHandler(err)
-			}
-			if err != nil {
-				return nil, err
-			}
-		}
-	// If callback
-	case ofoperation.CallbackUrl != "":
-		fmt.Printf("[Request `%s`] Executing callback `%s`\n",
-			of.reqId, ofoperation.CallbackUrl)
-		err = executeCallback(ofoperation, data)
-		if err != nil {
-			err = fmt.Errorf("Callback(%s), error: callback failed, %v",
-				ofoperation.CallbackUrl, err)
-			if ofoperation.FailureHandler != nil {
-				err = ofoperation.FailureHandler(err)
-			}
-			if err != nil {
-				return nil, err
-			}
-		}
-
-	// If modifier
-	default:
-		fmt.Printf("[Request `%s`] Executing modifier\n", of.reqId)
-		result, err = ofoperation.Mod(data)
-		if err != nil {
-			err = fmt.Errorf("error: Failed at modifier, %v", err)
-			return nil, err
-		}
-		if result == nil {
-			result = []byte("")
-		}
-	}
-
-	return result, nil
-}
-
 func (of *openFaasExecutor) HandleNextNode(partial *executor.PartialState) error {
 
 	state, err := partial.Encode()
@@ -198,6 +141,7 @@ func (of *openFaasExecutor) HandleNextNode(partial *executor.PartialState) error
 	httpreq.Header.Add("Accept", "application/json")
 	httpreq.Header.Add("Content-Type", "application/json")
 	httpreq.Header.Add("X-Faas-Flow-Reqid", of.reqId)
+	httpreq.Header.Set("X-Faas-Flow-State", "partial")
 
 	// extend req span for async call
 	of.tracer.extendReqSpan(of.reqId, of.openFaasEventHandler.currentNodeId,
@@ -218,6 +162,14 @@ func (of *openFaasExecutor) HandleNextNode(partial *executor.PartialState) error
 	return nil
 }
 
+func (of *openFaasExecutor) GetExecutionOption(operation sdk.Operation) map[string]interface{} {
+	options := make(map[string]interface{})
+	options["gateway"] = of.gateway
+	options["request-id"] = of.reqId
+
+	return options
+}
+
 // Executor
 
 func (of *openFaasExecutor) Configure(requestId string) {
@@ -399,7 +351,6 @@ func (of *openFaasExecutor) Handle(req *HttpRequest, response *HttpResponse) err
 		}
 		response.Body = resp
 		response.Header.Set("X-Faas-Flow-Reqid", of.reqId)
-		response.Header.Set("X-Faas-Flow-State", "partial")
 	}
 
 	response.StatusCode = http.StatusOK