Test resuming workflow

Author: grussorusso

internal/test/workflow_integration_test.go

@@ -114,7 +114,9 @@ func TestInvokeFC(t *testing.T) {
 
 	// check result
 	output := resultMap.Result[f.Signature.GetOutputs()[0].Name]
-	u.AssertEquals(t, length, output.(int))
+	if length != int(output.(float64)) {
+		t.FailNow()
+	}
 
 	// cleaning up function composition and function
 	err3 := wflow.Delete()
@@ -564,3 +566,52 @@ func TestInvokeWorkflowPassDoNothing(t *testing.T) {
 	u.AssertNilMsg(t, err, "Result not found")
 	u.AssertEquals(t, 3, result)
 }
+
+// TestResumeWorkflow offloads the execution of an invocation request to the same node
+func TestResumeWorkflow(t *testing.T) {
+
+	if testing.Short() {
+		t.Skip("Skipping integration test")
+	}
+
+	workflowName := "test"
+	// CREATE - we create a test function composition
+	length := 5
+	f, fArr, err := initializeSameFunctionSlice(length, "js")
+	u.AssertNil(t, err)
+	wflow, err := CreateSequenceWorkflow(fArr...)
+	wflow.Name = workflowName
+	u.AssertNil(t, err)
+	err1 := wflow.Save()
+	u.AssertNil(t, err1)
+
+	// INVOKE - we call the function composition
+	params := make(map[string]interface{})
+	params[f.Signature.GetInputs()[0].Name] = 0
+
+	request := workflow.NewRequest(shortuuid.New(), wflow, params)
+
+	progress := workflow.InitProgress(workflow.ReqId(request.Id), wflow)
+	pd := workflow.NewPartialData(workflow.ReqId(request.Id), wflow.Start.Next, "", request.Params)
+
+	err = workflow.SaveProgress(progress, true)
+	u.AssertNil(t, err)
+	err = workflow.SavePartialData(pd, true)
+	u.AssertNil(t, err)
+
+	resumedRequest := workflow.NewRequest(request.Id, wflow, params)
+	resumedRequest.Resuming = true
+
+	resultMap, err2 := wflow.Invoke(resumedRequest)
+	u.AssertNil(t, err2)
+
+	// check result
+	output := resultMap.Result[f.Signature.GetOutputs()[0].Name]
+	if length != int(output.(float64)) {
+		t.FailNow()
+	}
+
+	// cleaning up function composition and function
+	err3 := wflow.Delete()
+	u.AssertNil(t, err3)
+}

internal/workflow/partial_data.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"sync"
 	"time"
 
@@ -230,7 +231,10 @@ func savePartialDataToEtcd(pd *PartialData) error {
 	pdEtcdMutex.Lock()
 	defer pdEtcdMutex.Unlock()
 	// saves the json object into etcd
-	_, err = cli.Put(ctx, getPartialDataEtcdKey(pd.ReqId, pd.ForTask), string(payload))
+	key := getPartialDataEtcdKey(pd.ReqId, pd.ForTask)
+	log.Printf("Saving PD on etcd with key: %s\n", key)
+
+	_, err = cli.Put(ctx, key, string(payload))
 	if err != nil {
 		return fmt.Errorf("failed etcd Put partial data: %v", err)
 	}

internal/workflow/progress.go

@@ -207,7 +207,7 @@ func InitProgress(reqId ReqId, workflow *Workflow) *Progress {
 		ReadyToExecute: make([]TaskId, 0),
 	}
 
-	p.ReadyToExecute = append(p.ReadyToExecute, workflow.Start.GetId())
+	p.ReadyToExecute = append(p.ReadyToExecute, workflow.Start.Next)
 
 	return p
 }

internal/workflow/workflow.go

@@ -452,7 +452,7 @@ func (workflow *Workflow) Invoke(r *Request) (ExecutionReport, error) {
 
 		pd, err = RetrieveSinglePartialData(requestId, progress.ReadyToExecute[0], true)
 		if err != nil {
-			return ExecutionReport{}, fmt.Errorf("workflow resumed but unable to retrieve partial data of next task: %v", requestId)
+			return ExecutionReport{}, fmt.Errorf("workflow resumed but unable to retrieve partial data of next task: %v", progress.ReadyToExecute[0])
 		}
 	}