Completed workflow offloading mechanism

Author: grussorusso

internal/api/server.go

@@ -31,6 +31,7 @@ func StartAPIServer(e *echo.Echo) {
 	e.GET("/status", GetServerStatus)
 	// Workflow routes
 	e.POST("/workflow/invoke/:workflow", InvokeWorkflow)
+	e.POST("/workflow/resume/:workflow", ResumeWorkflow)
 	e.POST("/workflow/create", CreateWorkflowFromASL)
 	e.POST("/workflow/import", CreateWorkflow)
 	e.POST("/workflow/delete", DeleteWorkflow)

internal/api/workflow.go

@@ -142,6 +142,36 @@ func DeleteWorkflow(c echo.Context) error {
 	return c.JSON(http.StatusOK, response)
 }
 
+// ResumeWorkflow handles a workflow invocation resume request (workflow offloading).
+func ResumeWorkflow(e echo.Context) error {
+	workflowName := e.Param("workflow")
+	wflow, ok := workflow.Get(workflowName)
+	if !ok {
+		log.Printf("Dropping request for unknown workflow '%s'", workflowName)
+		return e.JSON(http.StatusNotFound, "function workflow '"+workflowName+"' does not exist")
+	}
+
+	var clientReq client.WorkflowInvocationResumeRequest
+	err := json.NewDecoder(e.Request().Body).Decode(&clientReq)
+	if err != nil && err != io.EOF {
+		log.Printf("Could not parse invoke request - error during decoding: %v", err)
+		return e.JSON(http.StatusInternalServerError, "failed to parse workflow invocation request. Check parameters and workflow definition")
+	}
+
+	req := workflowInvocationRequestPool.Get().(*workflow.Request)
+	req.W = wflow
+	req.Params = clientReq.Params
+	req.Arrival = time.Now()
+	req.QoS = clientReq.QoS
+	req.CanDoOffloading = clientReq.CanDoOffloading
+	req.Async = clientReq.Async
+	req.Resuming = true
+	req.Id = clientReq.ReqId
+	req.ExecReport.Reports = map[string]*function.ExecutionReport{}
+
+	return handleWorkflowInvocation(e, req)
+}
+
 // InvokeWorkflow handles a function workflow invocation request.
 func InvokeWorkflow(e echo.Context) error {
 	workflowName := e.Param("workflow")
@@ -165,10 +195,15 @@ func InvokeWorkflow(e echo.Context) error {
 	req.QoS = clientReq.QoS
 	req.CanDoOffloading = clientReq.CanDoOffloading
 	req.Async = clientReq.Async
-
+	req.Resuming = false
 	req.Id = fmt.Sprintf("%v-%s%d", wflow.Name, node.NodeIdentifier[len(node.NodeIdentifier)-5:], req.Arrival.Nanosecond())
 	req.ExecReport.Reports = map[string]*function.ExecutionReport{}
 
+	return handleWorkflowInvocation(e, req)
+}
+
+func handleWorkflowInvocation(e echo.Context, req *workflow.Request) error {
+
 	if req.Async {
 		go func() {
 			_, errInvoke := req.W.Invoke(req)
@@ -194,7 +229,7 @@ func InvokeWorkflow(e echo.Context) error {
 	}
 
 	// Synchronous execution of the workflow
-	_, err = req.W.Invoke(req)
+	_, err := req.W.Invoke(req)
 
 	defer workflowInvocationRequestPool.Put(req)
 

internal/cli/cli.go

@@ -470,11 +470,8 @@ func invokeWorkflow(cmd *cobra.Command, args []string) {
 	}
 
 	// Prepare request // TODO: it's ok to reuse the same type that function invocation uses?
-	request := client.InvocationRequest{
-		Params:   paramsMap,
-		QoSClass: api.DecodeServiceClass(qosClass),
-		// QoSClass:        qosClass,
-		QoSMaxRespT:     qosMaxRespT,
+	request := client.WorkflowInvocationRequest{
+		Params:          paramsMap,
 		CanDoOffloading: true,
 		Async:           asyncInvocation}
 	invocationBody, err := json.Marshal(request)

internal/client/types.go

@@ -26,13 +26,12 @@ type WorkflowInvocationRequest struct {
 	QoS             function.RequestQoS
 	CanDoOffloading bool
 	Async           bool
-	// NextNodes       []string // DagNodeId
-	// we do not add Progress here, only the next group of node that should execute
-	// in case of choice node, we retrieve the progress for each taskId and execute only the one that is not in Skipped State
-	// in case of fan out node, we retrieve all the progress and execute concurrently all the tasks in the group.
-	// in case of fan in node, we retrieve periodically all the progress of the previous nodes and start the merging only when all previous node are completed.
-	//   or simply, we can get the N partialData for the Fan Out, coming from the previous nodes.
-	//   furthermore, we should be careful not to run multiple fanIn at the same time!
+}
+
+// WorkflowInvocationResumeRequest is a request to resume the execution of a workflow (typically on a remote node)
+type WorkflowInvocationResumeRequest struct {
+	ReqId string
+	WorkflowInvocationRequest
 }
 
 type WorkflowCreationRequest struct {

internal/workflow/offloading_policy.go

@@ -0,0 +1,33 @@
+package workflow
+
+type OffloadingDecision struct {
+	Offload    bool   `json:"offload"`
+	RemoteHost string `json:"remote_host"`
+}
+
+type OffloadingPolicy interface {
+	Evaluate(r *Request, p *Progress) (OffloadingDecision, error)
+}
+
+type SimpleOffloadingPolicy struct{}
+
+func (policy *SimpleOffloadingPolicy) Evaluate(r *Request, p *Progress) (OffloadingDecision, error) {
+
+	completed := 0
+
+	if p == nil || !r.CanDoOffloading || len(p.ReadyToExecute) == 0 {
+		return OffloadingDecision{Offload: false}, nil
+	}
+
+	for _, s := range p.Status {
+		if s == Executed {
+			completed++
+		}
+	}
+
+	if completed >= 2 {
+		return OffloadingDecision{true, "127.0.0.1:1323"}, nil
+	}
+
+	return OffloadingDecision{Offload: false}, nil
+}

internal/workflow/progress.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"sync"
 	"time"
 
@@ -283,6 +284,7 @@ func RetrieveProgress(reqId ReqId, tryFromEtcd bool) (*Progress, bool) {
 		// cache miss - retrieve progress from ETCD
 		progress, err = getProgressFromEtcd(reqId)
 		if err != nil {
+			log.Printf("failed to retrieve progress from Etcd %v: %v", reqId, err)
 			return nil, false
 		}
 		// insert a new element to the cache
@@ -345,6 +347,7 @@ func saveProgressToEtcd(p *Progress) error {
 	}
 	// saves the json object into etcd
 	key := getProgressEtcdKey(p.ReqId)
+	log.Printf("Saving progress with key: %s", key)
 	progressMutexEtcd.Lock()
 	defer progressMutexEtcd.Unlock()
 	_, err = cli.Put(ctx, key, string(payload))

internal/workflow/workflow.go

@@ -5,6 +5,9 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/serverledge-faas/serverledge/internal/client"
+	"io"
+	"net/http"
 	"sort"
 	"time"
 
@@ -17,8 +20,7 @@ import (
 	"github.com/serverledge-faas/serverledge/internal/types"
 )
 
-// used to send output from parallel tasks to fan in task or to the next task
-// var outputChannel = make(chan map[string]interface{})
+var offloadingPolicy OffloadingPolicy = &SimpleOffloadingPolicy{} // TODO: handle initialization elsewhere
 
 // Workflow is a Workflow to drive the execution of the workflow
 type Workflow struct {
@@ -456,22 +458,90 @@ func (workflow *Workflow) Invoke(r *Request) (ExecutionReport, error) {
 
 	shouldContinue := true
 	for shouldContinue {
-		// TODO: introduce a workflow offloading policy
+		decision, err := offloadingPolicy.Evaluate(r, progress)
+		if err == nil && decision.Offload {
+			err = offload(r, decision.RemoteHost, progress, pd)
+			if err != nil {
+				return ExecutionReport{}, err
+			}
+			shouldContinue = false
+		} else {
+			pd, progress, shouldContinue, err = workflow.Execute(r, pd, progress)
+			if err != nil {
+				return ExecutionReport{}, fmt.Errorf("failed workflow execution: %v", err)
+			}
 
-		// TODO: if local execution:
-		// executing workflow
-		pd, progress, shouldContinue, err = workflow.Execute(r, pd, progress)
-		if err != nil {
-			return ExecutionReport{}, fmt.Errorf("failed workflow execution: %v", err)
+			if !shouldContinue {
+				r.ExecReport.Result = pd.Data
+			}
 		}
-	}
 
-	r.ExecReport.Result = pd.Data
+	}
 
 	// TODO: remove r.ExecReport
 	return r.ExecReport, nil
 }
 
+func offload(r *Request, hostPort string, progress *Progress, pd *PartialData) error {
+
+	err := saveProgressToEtcd(progress)
+	if err != nil {
+		return fmt.Errorf("Could not save progress: %v", err)
+	}
+	err = savePartialDataToEtcd(pd)
+	if err != nil {
+		return fmt.Errorf("Could not save partial data: %v", err)
+	}
+
+	request := client.WorkflowInvocationResumeRequest{
+		ReqId: r.Id,
+		WorkflowInvocationRequest: client.WorkflowInvocationRequest{
+			Params:          r.Params,
+			CanDoOffloading: false,
+			Async:           r.Async,
+		},
+	}
+	invocationBody, err := json.Marshal(request)
+	if err != nil {
+		return fmt.Errorf("JSON marshaling failed: %v", err)
+	}
+
+	// Send invocation request
+	url := fmt.Sprintf("http://%s/workflow/resume/%s", hostPort, r.W.Name)
+	resp, err := utils.PostJson(url, invocationBody)
+	if err != nil {
+		return fmt.Errorf("HTTP request for offloading failed: %v", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("failed offloaded workflow: %v", err)
+	}
+
+	if r.Async {
+		// no need to parse and return response
+		return nil
+	}
+
+	var response InvocationResponse
+	body, _ := io.ReadAll(resp.Body)
+	err = json.Unmarshal(body, &response)
+	if err != nil {
+		return fmt.Errorf("Failed InvocationResponse unmarshaling: %v", err)
+	}
+
+	if !response.Success {
+		return fmt.Errorf("failed offloaded workflow: %v", err)
+	}
+
+	r.ExecReport.Result = response.Result
+
+	for k, v := range response.Reports {
+		r.ExecReport.Reports[k] = v
+	}
+
+	return nil
+}
+
 // Delete removes the Workflow from cache and from etcd, so it cannot be invoked anymore
 func (workflow *Workflow) Delete() error {
 	cli, err := utils.GetEtcdClient()