Refactor template to separate packages, extract request handlers and add REST api

Author: pasdam

template/faas-flow/config/consul_dc.go

@@ -0,0 +1,13 @@
+package config
+
+import (
+	"os"
+)
+
+func ConsulDC() string {
+	val := os.Getenv("consul_dc")
+	if len(val) == 0 {
+		val = "dc1"
+	}
+	return val
+}

template/faas-flow/config/consul_url.go

@@ -0,0 +1,13 @@
+package config
+
+import (
+	"os"
+)
+
+func ConsulURL() string {
+	val := os.Getenv("consul_url")
+	if len(val) == 0 {
+		val = "consul.faasflow:8500"
+	}
+	return val
+}

template/faas-flow/config/gateway_url.go

@@ -0,0 +1,14 @@
+package config
+
+import (
+	"os"
+)
+
+// GatewayURL return the gateway address from env
+func GatewayURL() string {
+	gateway := os.Getenv("gateway")
+	if gateway == "" {
+		gateway = "gateway.openfaas:8080"
+	}
+	return gateway
+}

template/faas-flow/config/parse_int_or_duration_value.go

@@ -0,0 +1,21 @@
+package config
+
+import (
+	"strconv"
+	"time"
+)
+
+func parseIntOrDurationValue(val string, fallback time.Duration) time.Duration {
+	if len(val) > 0 {
+		parsedVal, parseErr := strconv.Atoi(val)
+		if parseErr == nil && parsedVal >= 0 {
+			return time.Duration(parsedVal) * time.Second
+		}
+	}
+
+	duration, durationErr := time.ParseDuration(val)
+	if durationErr != nil {
+		return fallback
+	}
+	return duration
+}

template/faas-flow/config/read_timeout.go

@@ -0,0 +1,10 @@
+package config
+
+import (
+	"os"
+	"time"
+)
+
+func ReadTimeout() time.Duration {
+	return parseIntOrDurationValue(os.Getenv("read_timeout"), 10*time.Second)
+}

template/faas-flow/config/trace_server.go

@@ -0,0 +1,14 @@
+package config
+
+import (
+	"os"
+)
+
+// TraceServer get the traceserver address
+func TraceServer() string {
+	traceServer := os.Getenv("trace_server")
+	if traceServer == "" {
+		traceServer = "jaeger.faasflow:5775"
+	}
+	return traceServer
+}

template/faas-flow/config/write_timeout.go

@@ -0,0 +1,10 @@
+package config
+
+import (
+	"os"
+	"time"
+)
+
+func WriteTimeout() time.Duration {
+	return parseIntOrDurationValue(os.Getenv("write_timeout"), 10*time.Second)
+}

template/faas-flow/executor/execute_flow_handler.go

@@ -0,0 +1,65 @@
+package executor
+
+import (
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+
+	"github.com/s8sg/faas-flow/sdk/executor"
+)
+
+func executeFlowHandler(w http.ResponseWriter, req *http.Request, id string, ex executor.Executor) ([]byte, error) {
+	log.Printf("Requested %s %s\n", req.Method, req.URL)
+	log.Printf("Executing flow %s\n", id)
+
+	var stateOption executor.ExecutionStateOption
+
+	body, err := ioutil.ReadAll(req.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO: fix this
+	openFaasEx := ex.(*openFaasExecutor)
+
+	state := req.Header.Get("X-Faas-Flow-State")
+	callbackURL := req.Header.Get("X-Faas-Flow-Callback-Url")
+	openFaasEx.callbackURL = callbackURL
+	if state == "" {
+		rawRequest := &executor.RawRequest{}
+		rawRequest.Data = body
+		rawRequest.Query = req.URL.RawQuery
+		rawRequest.AuthSignature = req.Header.Get("X-Hub-Signature")
+		// Check if any request Id is passed
+		if id != "" {
+			rawRequest.RequestId = id
+		}
+		stateOption = executor.NewRequest(rawRequest)
+
+	} else {
+		if id == "" {
+			return nil, errors.New("request ID not set in partial request")
+		}
+
+		openFaasEx.openFaasEventHandler.header = req.Header
+		partialState, err := executor.DecodePartialReq(body)
+		if err != nil {
+			return nil, errors.New("failed to decode partial state")
+		}
+		stateOption = executor.PartialRequest(partialState)
+	}
+
+	// Create a flow executor, OpenFaaSExecutor implements executor
+	flowExecutor := executor.CreateFlowExecutor(ex, nil)
+	resp, err := flowExecutor.Execute(stateOption)
+	if err != nil {
+		return nil, fmt.Errorf("failed to execute request. %s", err.Error())
+	}
+	headers := w.Header()
+	headers["X-Faas-Flow-Reqid"] = []string{flowExecutor.GetReqId()}
+	headers["X-Faas-Flow-Callback-Url"] = []string{callbackURL}
+
+	return resp, nil
+}

template/faas-flow/executor/executor.go

@@ -0,0 +1,8 @@
+package executor
+
+import "github.com/s8sg/faas-flow/sdk"
+
+var (
+	stateStore sdk.StateStore
+	dataStore  sdk.DataStore
+)

template/faas-flow/executor/flow_state_handler.go

@@ -0,0 +1,22 @@
+package executor
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/s8sg/faas-flow/sdk/executor"
+)
+
+func flowStateHandler(w http.ResponseWriter, req *http.Request, id string, ex executor.Executor) ([]byte, error) {
+	log.Printf("Get flow state: %s\n", id)
+
+	flowExecutor := executor.CreateFlowExecutor(ex, nil)
+	state, err := flowExecutor.GetState(id)
+	if err != nil {
+		log.Printf(err.Error())
+		return nil, fmt.Errorf("failed to get request state for %s, check if request is active", id)
+	}
+
+	return []byte(state), nil
+}