Improved Progress/TaskData saving

Author: grussorusso

internal/config/keys.go

@@ -87,6 +87,8 @@ const TRACING_ENABLED = "tracing.enabled"
 // Custom output file for traces
 const TRACING_OUTFILE = "tracing.outfile"
 
+const WORKFLOW_ALWAYS_SAVE_PROGRESS = "workflow.always_save_progress"
+
 // Workflow offloading policy to use
 // Possible values: "disable", "ilp"
 const WORKFLOW_OFFLOADING_POLICY = "workflow.offloading.policy"

internal/workflow/progress.go

@@ -193,7 +193,7 @@ func (p *Progress) Save() error {
 	}
 	// saves the json object into etcd
 	key := getProgressEtcdKey(p.ReqId)
-	log.Printf("Saving progress with key: %s", key)
+	log.Printf("[Rq-%v] Saving progress - bytes: %d", p.ReqId, len(payload))
 
 	_, err = cli.Put(ctx, key, string(payload))
 	if err != nil {

internal/workflow/task_data.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"log"
 	"time"
 
 	"github.com/serverledge-faas/serverledge/utils"
@@ -60,17 +61,19 @@ func (td *TaskData) Save(reqId ReqId, task TaskId) error {
 		return err
 	}
 	ctx := context.TODO()
-	// marshal the progress object into json
+	// marshal task data object into json
 	payload, err := json.Marshal(td)
 	if err != nil {
-		return fmt.Errorf("could not marshal progress: %v", err)
+		return fmt.Errorf("could not marshal task data: %v", err)
 	}
 	// saves the json object into etcd
 	key := getTaskDataEtcdKey(reqId, task)
 
+	log.Printf("[Rq-%v] Saving task data to etcd - key: %s - bytes: %d", reqId, key, len(payload))
 	_, err = cli.Put(ctx, key, string(payload))
 	if err != nil {
-		utils.TryEtcdReconnection()
+		log.Printf("[Rq-%v] Could not save task data due to failed Put()...retrying Etcd connection: %v", reqId, err)
+		utils.TriggerEtcdReconnection()
 		return fmt.Errorf("failed etcd Put partial data: %v", err)
 	}
 	return nil
@@ -86,7 +89,7 @@ func RetrievePartialData(reqId ReqId, task TaskId) (*TaskData, error) {
 	key := getTaskDataEtcdKey(reqId, task)
 	getResponse, err := cli.Get(ctx, key)
 	if err != nil {
-		utils.TryEtcdReconnection()
+		utils.TriggerEtcdReconnection()
 		return nil, fmt.Errorf("failed to retrieve PD for requestId %s: %v", key, err)
 	}
 	if len(getResponse.Kvs) < 1 {

internal/workflow/workflow.go

@@ -425,6 +425,8 @@ func (wflow *Workflow) savePartialDataForReadyTasks(requestId ReqId, progress *P
 // Invoke schedules each function of the workflow and invokes them
 func (wflow *Workflow) Invoke(r *Request) error {
 
+	alwaysSaveProgress := config.GetBool(config.WORKFLOW_ALWAYS_SAVE_PROGRESS, false)
+
 	var err error
 	requestId := ReqId(r.Id)
 
@@ -437,9 +439,11 @@ func (wflow *Workflow) Invoke(r *Request) error {
 	dataMap := make(map[TaskId]*TaskData)
 
 	if len(progress.ReadyToExecute) == 0 {
-		return fmt.Errorf("wflow resumed but no task is ready for execution: %v", requestId)
+		return fmt.Errorf("[Rq-%v] wflow resumed but no task is ready for execution", requestId)
 	}
 
+	log.Printf("[Rq-%v] Starting/resuming execution (%d to executed)", requestId, len(progress.ReadyToExecute))
+
 	for len(progress.ReadyToExecute) > 0 {
 		t0 := time.Now()
 		decision, err := offloadingPolicy.Evaluate(r, progress)
@@ -450,7 +454,7 @@ func (wflow *Workflow) Invoke(r *Request) error {
 			return fmt.Errorf("an error occurred in policy evaluation: %v", err)
 		}
 
-		if decision.Offload {
+		if decision.Offload || alwaysSaveProgress {
 			err := progress.Save()
 			if err != nil {
 				return fmt.Errorf("Could not save progress: %v", err)
@@ -462,24 +466,26 @@ func (wflow *Workflow) Invoke(r *Request) error {
 				return fmt.Errorf("Could not save partial data: %v", err)
 			}
 
-			log.Printf("Offloading request: %v", requestId)
+		}
 
+		if decision.Offload {
 			err = offload(r, &decision)
 			if err != nil {
 				return err
 			}
 
 			if r.ExecReport.Result != nil {
 				// Workflow execution has completed on remote node
-				break
+				log.Printf("[Rq-%v] Workflow has completed on remote node", requestId)
+				return nil
 			}
 
 			progress, err = RetrieveProgress(requestId)
 			if err != nil {
 				return fmt.Errorf("Could not retrieve progress after offloading: %v", err)
 			}
 
-			log.Printf("Ready to execute after offloading: %v", progress.ReadyToExecute)
+			log.Printf("[Rq-%v] Ready to execute after offloading: %v", requestId, progress.ReadyToExecute)
 		} else {
 			// pick next executable task
 			var taskToExecute TaskId = ""
@@ -489,9 +495,12 @@ func (wflow *Workflow) Invoke(r *Request) error {
 				}
 			}
 			if taskToExecute == "" {
+				log.Printf("[Rq-%v] Workflow has not completed but there is nothing left to execute in the plan", requestId)
 				break
 			}
 
+			log.Printf("[Rq-%v] Now going to execute %s", requestId, taskToExecute)
+
 			// Prepare input for taskToExecute
 			var input *TaskData
 			if wflow.Tasks[taskToExecute].GetType() == Start {
@@ -530,11 +539,15 @@ func (wflow *Workflow) Invoke(r *Request) error {
 				}
 			}
 
+			log.Printf("[Rq-%v] Executed %s", requestId, taskToExecute)
+
 			dataMap[taskToExecute] = output
 
 			if len(progress.ReadyToExecute) == 0 && output != nil {
 				r.ExecReport.Result = output.Data
 
+				log.Printf("[Rq-%v] Workflow completed", requestId)
+
 				if isProgressOnEtcd {
 					err = DeleteProgress(requestId)
 					if err != nil {
@@ -568,7 +581,7 @@ func (wflow *Workflow) Invoke(r *Request) error {
 
 func offload(r *Request, policyDecision *OffloadingDecision) error {
 
-	log.Printf("Offloading decision: %v", policyDecision)
+	log.Printf("[Rq-%v] Offloading decision: %v", r.Id, policyDecision)
 
 	request := WorkflowInvocationResumeRequest{
 		ReqId: r.Id,