Added the ability to start on any node in a workflow

backend/go-app/main.go

@@ -242,9 +242,11 @@ type AppInfo struct {
 // May 2020: Reused for onprem schedules - Id, Seconds, WorkflowId and argument
 type ScheduleOld struct {
 	Id                   string       `json:"id" datastore:"id"`
+	StartNode            string       `json:"start_node" datastore:"start_node"`
 	Seconds              int          `json:"seconds" datastore:"seconds"`
 	WorkflowId           string       `json:"workflow_id" datastore:"workflow_id", `
 	Argument             string       `json:"argument" datastore:"argument"`
+	WrappedArgument      string       `json:"wrapped_argument" datastore:"wrapped_argument"`
 	AppInfo              AppInfo      `json:"appinfo" datastore:"appinfo,noindex"`
 	Finished             bool         `json:"finished" finished:"id"`
 	BaseAppLocation      string       `json:"base_app_location" datastore:"baseapplocation,noindex"`
@@ -6034,7 +6036,7 @@ func runInit(ctx context.Context) {
 			job := func() {
 				request := &http.Request{
 					Method: "POST",
-					Body:   ioutil.NopCloser(strings.NewReader(schedule.Argument)),
+					Body:   ioutil.NopCloser(strings.NewReader(schedule.WrappedArgument)),
 				}
 
 				_, _, err := handleExecution(schedule.WorkflowId, Workflow{}, request)
@@ -6213,7 +6215,6 @@ func init() {
 	r.HandleFunc("/api/v1/workflows/queue/confirm", handleGetWorkflowqueueConfirm).Methods("POST")
 	r.HandleFunc("/api/v1/workflows/schedules", handleGetSchedules).Methods("GET", "OPTIONS")
 	r.HandleFunc("/api/v1/workflows/download_remote", loadSpecificWorkflows).Methods("POST", "OPTIONS")
-	//r.HandleFunc("/api/v1/workflows/{key}/execute_fs", executeWorkflowFS)
 	r.HandleFunc("/api/v1/workflows/{key}/execute", executeWorkflow).Methods("GET", "POST", "OPTIONS")
 	r.HandleFunc("/api/v1/workflows/{key}/schedule", scheduleWorkflow).Methods("POST", "OPTIONS")
 	r.HandleFunc("/api/v1/workflows/{key}/schedule/{schedule}", stopSchedule).Methods("DELETE", "OPTIONS")

backend/go-app/walkoff.go

@@ -255,7 +255,8 @@ type Workflow struct {
 	Triggers      []Trigger  `json:"triggers" datastore:"triggers,noindex"`
 	Schedules     []Schedule `json:"schedules" datastore:"schedules,noindex"`
 	Configuration struct {
-		ExitOnError bool `json:"exit_on_error" datastore:"exit_on_error"`
+		ExitOnError  bool `json:"exit_on_error" datastore:"exit_on_error"`
+		StartFromTop bool `json:"start_from_top" datastore:"start_from_top"`
 	} `json:"configuration,omitempty" datastore:"configuration"`
 	Errors            []string `json:"errors,omitempty" datastore:"errors"`
 	Tags              []string `json:"tags,omitempty" datastore:"tags"`
@@ -414,7 +415,7 @@ func getWorkflowQueue(ctx context.Context, id string) (ExecutionRequestWrapper,
 //}
 
 // Frequency = cronjob OR minutes between execution
-func createSchedule(ctx context.Context, scheduleId, workflowId, name, frequency string, body []byte) error {
+func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode, frequency string, body []byte) error {
 	var err error
 	testSplit := strings.Split(frequency, "*")
 	cronJob := ""
@@ -448,10 +449,14 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, frequency
 	// FIXME:
 	// This may run multiple places if multiple servers,
 	// but that's a future problem
+	log.Printf("BODY: %s", string(body))
+	parsedArgument := strings.Replace(string(body), "\"", "\\\"", -1)
+	bodyWrapper := fmt.Sprintf(`{"start": "%s", "execution_argument": "%s"}`, startNode, parsedArgument)
+	log.Printf("WRAPPER BODY: \n%s", bodyWrapper)
 	job := func() {
 		request := &http.Request{
 			Method: "POST",
-			Body:   ioutil.NopCloser(strings.NewReader(string(body))),
+			Body:   ioutil.NopCloser(strings.NewReader(bodyWrapper)),
 		}
 
 		_, _, err := handleExecution(workflowId, Workflow{}, request)
@@ -475,7 +480,9 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, frequency
 	schedule := ScheduleOld{
 		Id:                   scheduleId,
 		WorkflowId:           workflowId,
+		StartNode:            startNode,
 		Argument:             string(body),
+		WrappedArgument:      bodyWrapper,
 		Seconds:              newfrequency,
 		CreationTime:         timeNow,
 		LastModificationtime: timeNow,
@@ -708,7 +715,23 @@ func findChildNodes(workflowExecution WorkflowExecution, nodeId string) []string
 		}
 	}
 
-	return allChildren
+	// Remove potential duplicates
+	newNodes := []string{}
+	for _, tmpnode := range allChildren {
+		found := false
+		for _, newnode := range newNodes {
+			if newnode == tmpnode {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			newNodes = append(newNodes, tmpnode)
+		}
+	}
+
+	return newNodes
 }
 
 func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
@@ -789,22 +812,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 			// Find underlying nodes and add them
 		} else {
 			// Finds ALL childnodes to set them to SKIPPED
-			tmpNodes := findChildNodes(*workflowExecution, actionResult.Action.ID)
+			childNodes = findChildNodes(*workflowExecution, actionResult.Action.ID)
-			for _, tmpnode := range tmpNodes {
-				found := false
-				for _, newnode := range childNodes {
-					if newnode == tmpnode {
-						found = true
-						break
-
-					}
-				}
-
-				if !found {
-					childNodes = append(childNodes, tmpnode)
-				}
-
-			}
 			// Remove duplicates
 			log.Printf("CHILD NODES: %d", len(childNodes))
 			for _, nodeId := range childNodes {
@@ -1568,25 +1576,27 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) {
 		resp.Write([]byte(`{"success": false}`))
 		return
 	}
-	//} else if err != nil {
-	//	log.Printf("Error getting item: %v", err)
-	//} else {
-	//	// FIXME - verify if value is ok? Can unmarshal etc.
-	//	err = json.Unmarshal(item.Value, &workflowApps)
-	//	if err != nil {
-	//		log.Printf("Failed unmarshaling allworkflowapps from memcache: %s", err)
-	//		resp.WriteHeader(401)
-	//		resp.Write([]byte(`{"success": false}`))
-	//		return
-	//	}
-
-	//	if userErr == nil && len(user.PrivateApps) > 0 {
-	//		workflowApps = append(workflowApps, user.PrivateApps...)
-	//	}
-	//}
 
 	// Started getting the single apps, but if it's weird, this is faster
-	log.Println("Apps set done")
+	// 1. Check workflow.Start
+	// 2. Check if any node has "isStartnode"
+	if len(workflow.Actions) > 0 {
+		index := -1
+		for indexFound, action := range workflow.Actions {
+			//log.Println("Apps set done")
+			if workflow.Start == action.ID {
+				index = indexFound
+			}
+		}
+
+		if index >= 0 {
+			workflow.Actions[0].IsStartNode = true
+		} else {
+			resp.WriteHeader(401)
+			resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "You need to set a startnode."}`)))
+			return
+		}
+	}
 
 	// Check every app action and param to see whether they exist
 	newActions = []Action{}
@@ -1982,7 +1992,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 		//log.Printf("Execution data: %#v", execution)
 		if len(execution.Start) == 36 {
 			log.Printf("SHOULD START ON NODE %s", execution.Start)
-			workflow.Start = execution.Start
+			workflowExecution.Start = execution.Start
 
 			found := false
 			for _, action := range workflow.Actions {
@@ -1995,6 +2005,10 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 				log.Printf("ACTION %s WAS NOT FOUND!", workflow.Start)
 				return WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", workflow.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", workflow.Start))
 			}
+		} else if len(execution.Start) > 0 {
+
+			log.Printf("START ACTION %s IS WRONG ID LENGTH %d!", len(execution.Start))
+			return WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", execution.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", execution.Start))
 		}
 
 		if len(execution.ExecutionId) == 36 {
@@ -2080,9 +2094,10 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 			makeNew = false
 		}
 
+		// Don't override workflow defaults
 		if startok {
 			log.Printf("Setting start to %s based on query!", start[0])
-			workflowExecution.Workflow.Start = start[0]
+			//workflowExecution.Workflow.Start = start[0]
 			workflowExecution.Start = start[0]
 		}
 
@@ -2134,9 +2149,18 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 	//}
 
 	//log.Println(string(mappedData))
+
+	log.Printf("STARTNODE: %s", workflowExecution.Start)
+	if len(workflowExecution.Start) == 0 {
+		workflowExecution.Start = workflowExecution.Workflow.Start
+	}
+
+	childNodes := findChildNodes(workflowExecution, workflowExecution.Start)
+
 	topic := "workflows"
 	// FIXME - remove this?
 	newActions := []Action{}
+	defaultResults := []ActionResult{}
 	for _, action := range workflowExecution.Workflow.Actions {
 		action.LargeImage = ""
 		//log.Println(action.Environment)
@@ -2144,15 +2168,47 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 		if action.Environment == "" {
 			return WorkflowExecution{}, fmt.Sprintf("Environment is not defined for %s", action.Name), errors.New("Environment not defined!")
 		}
-		newActions = append(newActions, action)
-	}
-	workflowExecution.Workflow.Actions = newActions
 
-	//log.Printf("%#v", workflowExecution.Workflow.Actions)
+		newActions = append(newActions, action)
+
+		// If the node is NOT found, it's supposed to be set to SKIPPED,
+		// as it's not a childnode of the startnode
+		// This is a configuration item for the workflow itself.
+		if !workflowExecution.Workflow.Configuration.StartFromTop {
+			found := false
+			for _, nodeId := range childNodes {
+				if nodeId == action.ID {
+					//log.Printf("Found %s", action.ID)
+					found = true
+				}
+			}
+
+			if !found {
+				if action.ID == workflowExecution.Start {
+					continue
+				}
+
+				log.Printf("Should set %s to SKIPPED as it's NOT a childnode of the startnode.", action.ID)
+				defaultResults = append(defaultResults, ActionResult{
+					Action:        action,
+					ExecutionId:   workflowExecution.ExecutionId,
+					Authorization: workflowExecution.Authorization,
+					Result:        "Skipped because it's not under the startnode",
+					StartedAt:     0,
+					CompletedAt:   0,
+					Status:        "SKIPPED",
+				})
+			}
+		}
+	}
 
 	// Verification for execution environments
+	workflowExecution.Results = defaultResults
+	workflowExecution.Workflow.Actions = newActions
 	onpremExecution := false
 	environments := []string{}
+
+	// Check if the actions are children of the startnode?
 	for _, action := range workflowExecution.Workflow.Actions {
 		if action.Environment != cloudname {
 			found := false
@@ -2454,7 +2510,7 @@ func deleteSchedule(ctx context.Context, id string) error {
 		return err
 	} else {
 		if value, exists := scheduledJobs[id]; exists {
-			log.Printf("STOP THIS ONE: %s", value)
+			log.Printf("STOPPING THIS SCHEDULE: %s", id)
 			// Looks like this does the trick? Hurr
 			value.Lock()
 		} else {
@@ -2567,6 +2623,20 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) {
 		return
 	}
 
+	// Finds the startnode for the specific schedule
+	startNode := ""
+	for _, branch := range workflow.Branches {
+		if branch.SourceID == schedule.Id {
+			startNode = branch.DestinationID
+		}
+	}
+
+	if startNode == "" {
+		startNode = workflow.Start
+	}
+
+	log.Printf("Startnode: %s", startNode)
+
 	if len(schedule.Id) != 36 {
 		log.Printf("ID length is not 36 for schedule: %s", err)
 		resp.WriteHeader(http.StatusInternalServerError)
@@ -2612,6 +2682,7 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) {
 		schedule.Id,
 		workflow.ID,
 		schedule.Name,
+		startNode,
 		schedule.Frequency,
 		[]byte(parsedBody),
 	)

frontend/src/AngularWorkflow.js

@@ -795,14 +795,14 @@ const AngularWorkflow = (props) => {
 			setSelectedAction({})
 			setSelectedTrigger({})
 		} else {
-			alert.info("Can't edit branches from triggers") 
+			//alert.info("Can't edit branches from triggers") 
 		}
 	}
 
 	const onNodeSelect = (event) => {
 		const data = event.target.data()
 		setLastSaved(false)
-		//console.log(data)
+		console.log(data)
 
 		if (data.type === "ACTION") {
 			// FIXME - unselect
@@ -968,7 +968,7 @@ const AngularWorkflow = (props) => {
 					alert.success("Changed startnode to "+ele.data()["label"])
 					ele.data("isStartNode", true)
 					workflow.start = ele.id()
-					return
+					return true
 				}
 			});
 		}
@@ -2555,33 +2555,40 @@ const AngularWorkflow = (props) => {
 		if (oldstartnode.length > 0) {
 			oldstartnode[0].data("isStartNode", false)
 			var oldnodecnt = workflow.actions.findIndex(a => a.id === workflow.start)
-			workflow.actions[oldnodecnt].isStartNode = false
+			if (workflow.actions[oldnodecnt] !== undefined) {
+				workflow.actions[oldnodecnt].isStartNode = false
+			}
 		}
 
 		var newstartnode = cy.getElementById(selectedAction.id)
 		if (newstartnode.length > 0) {
 			newstartnode[0].data("isStartNode", true)
 			var newnodecnt = workflow.actions.findIndex(a => a.id === selectedAction.id)
-			workflow.actions[newnodecnt].isStartNode = true
+			console.log("NEW NODE CNT: ", newnodecnt)
+			if (workflow.actions[newnodecnt] !== undefined) {
+				workflow.actions[newnodecnt].isStartNode = true
+				console.log(workflow.actions[newnodecnt])
+			}
 		}
 
 		// Find branches with triggers as source nodes
 		// Move these targets to be the new node
 		// Set arrows pointing to new startnode with errors
-		for (var key in workflow.branches) {
+		//for (var key in workflow.branches) {
-			var item = workflow.branches[key]	
+		//	var item = workflow.branches[key]	
-			if (item.destination_id === oldstartnode[0].data()["id"]) {
+		//	if (item.destination_id === oldstartnode[0].data()["id"]) {
-				var curbranch = cy.getElementById(item.id)
+		//		var curbranch = cy.getElementById(item.id)
-				if (curbranch.length > 0) {
+		//		if (curbranch.length > 0) {
-					//console.log(curbranch[0].data())
+		//			//console.log(curbranch[0].data())
-					//curbranch[0].data("target", selectedAction.id)
+		//			//curbranch[0].data("target", selectedAction.id)
-					curbranch[0].data("hasErrors", true)
+		//			//curbranch[0].data("hasErrors", true)
-					//workflow.branches[key].destination_id = selectedAction.id
+		//			//workflow.branches[key].destination_id = selectedAction.id
-					//console.log(curbranch[0].data())
+		//			//console.log(curbranch[0].data())
-				}
+		//		}
-			}
+		//	}
-		}
+		//}
 
+		setUpdate("start_node"+selectedAction.id)
 		workflow.start = selectedAction.id
 		setWorkflow(workflow)
 		//setStartNode(selectedAction.id)

frontend/src/Workflows.js

@@ -875,6 +875,9 @@ const Workflows = (props) => {
 					<Button style={{}} disabled={newWorkflowName.length === 0} onClick={() => {
 						if (editingWorkflow.id !== undefined) {
 							setNewWorkflow(newWorkflowName, newWorkflowDescription, editingWorkflow, false)
+							setNewWorkflowName("")
+							setNewWorkflowDescription("")
+							setEditingWorkflow({})
 						} else {
 							setNewWorkflow(newWorkflowName, newWorkflowDescription, {}, true)
 						}

functions/onprem/worker/worker.go

@@ -57,22 +57,22 @@ type Org struct {
 	Id    string `json:"id"`
 }
 
-// FIXME: Generate a callback authentication ID?
 type WorkflowExecution struct {
-	Type               string         `json:"type"`
+	Type               string         `json:"type" datastore:"type"`
-	Status             string         `json:"status"`
+	Status             string         `json:"status" datastore:"status"`
-	ExecutionId        string         `json:"execution_id"`
+	Start              string         `json:"start" datastore:"start"`
-	ExecutionArgument  string         `json:"execution_argument"`
+	ExecutionArgument  string         `json:"execution_argument" datastore:"execution_argument"`
-	WorkflowId         string         `json:"workflow_id"`
+	ExecutionId        string         `json:"execution_id" datastore:"execution_id"`
-	LastNode           string         `json:"last_node"`
+	WorkflowId         string         `json:"workflow_id" datastore:"workflow_id"`
-	Authorization      string         `json:"authorization"`
+	LastNode           string         `json:"last_node" datastore:"last_node"`
-	Result             string         `json:"result"`
+	Authorization      string         `json:"authorization" datastore:"authorization"`
-	StartedAt          int64          `json:"started_at"`
+	Result             string         `json:"result" datastore:"result,noindex"`
-	CompletedAt        int64          `json:"completed_at"`
+	StartedAt          int64          `json:"started_at" datastore:"started_at"`
-	ProjectId          string         `json:"project_id"`
+	CompletedAt        int64          `json:"completed_at" datastore:"completed_at"`
-	Locations          []string       `json:"locations"`
+	ProjectId          string         `json:"project_id" datastore:"project_id"`
-	Workflow           Workflow       `json:"workflow"`
+	Locations          []string       `json:"locations" datastore:"locations"`
-	Results            []ActionResult `json:"results"`
+	Workflow           Workflow       `json:"workflow" datastore:"workflow,noindex"`
+	Results            []ActionResult `json:"results" datastore:"results,noindex"`
 	ExecutionVariables []struct {
 		Description string `json:"description" datastore:"description"`
 		ID          string `json:"id" datastore:"id"`
@@ -422,16 +422,39 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 	}
 
 	onpremApps := []string{}
-	startAction := workflowExecution.Workflow.Start
+	startAction := workflowExecution.Start
+	log.Printf("Startaction: %s", startAction)
 	toExecuteOnprem := []string{}
 	parents := map[string][]string{}
 	children := map[string][]string{}
 
-	// source = parent, dest = child
+	// source = parent node, dest = child node
 	// parent can have more children, child can have more parents
 	for _, branch := range workflowExecution.Workflow.Branches {
-		parents[branch.DestinationID] = append(parents[branch.DestinationID], branch.SourceID)
+		// Check what the parent is first. If it's trigger - skip
-		children[branch.SourceID] = append(children[branch.SourceID], branch.DestinationID)
+		sourceFound := false
+		destinationFound := false
+		for _, action := range workflowExecution.Workflow.Actions {
+			if action.ID == branch.SourceID {
+				sourceFound = true
+			}
+
+			if action.ID == branch.DestinationID {
+				destinationFound = true
+			}
+		}
+
+		if sourceFound {
+			parents[branch.DestinationID] = append(parents[branch.DestinationID], branch.SourceID)
+		} else {
+			log.Printf("ID %s was not found in actions! Skipping parent. (TRIGGER?)", branch.SourceID)
+		}
+
+		if destinationFound {
+			children[branch.SourceID] = append(children[branch.SourceID], branch.DestinationID)
+		} else {
+			log.Printf("ID %s was not found in actions! Skipping child. (TRIGGER?)", branch.SourceID)
+		}
 	}
 
 	log.Printf("Actions: %d", len(workflowExecution.Workflow.Actions))
@@ -483,12 +506,15 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 	// Process the parents etc. How?
 	visited := []string{}
 	executed := []string{}
-	nextActions := []string{}
+	nextActions := []string{startAction}
+	firstIteration := true
 	for {
 		queueNodes := []string{}
 
 		if len(workflowExecution.Results) == 0 {
 			nextActions = []string{startAction}
+		} else if firstIteration {
+			firstIteration = false
 		} else {
 			// This is to re-check the nodes that exist and whether they should continue
 			appendActions := []string{}
@@ -548,7 +574,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 		// care if it gets stuck in a loop.
 		// FIXME: Force killing a worker should result in a notification somewhere
 		if len(nextActions) == 0 {
-			log.Println("No next action. Finished?")
+			log.Printf("No next action. Finished? Result vs Actions: %d - %d", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions))
 			if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions) {
 				shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID)
 			}
@@ -569,7 +595,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 				}
 			}
 
-			log.Printf("SOMETHING IS MISSING!: %#v", notFound)
+			//log.Printf("SOMETHING IS MISSING!: %#v", notFound)
 			for _, item := range notFound {
 				if arrayContains(executed, item) {
 					log.Printf("%s has already executed but no result!", item)
@@ -615,6 +641,8 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 				}
 			}
 		}
+		//log.Printf("NEXT: %s", nextActions)
+		//log.Printf("queueNodes: %s", queueNodes)
 
 		// IF NOT VISITED && IN toExecuteOnPrem
 		// SKIP if it's not onprem
@@ -660,11 +688,12 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 			}
 
 			if continueOuter {
-				log.Printf("Parents of %s aren't finished: %s", nextAction, strings.Join(parents[nextAction], ", "))
+				//log.Printf("Parents of %s aren't finished: %s", nextAction, strings.Join(parents[nextAction], ", "))
-				for _, tmpaction := range parents[nextAction] {
+				//for _, tmpaction := range parents[nextAction] {
-					action := getAction(workflowExecution, tmpaction)
+				//	action := getAction(workflowExecution, tmpaction)
-					//log.Printf("Parent: %s", action.Label)
+				//	_ = action
-				}
+				//	//log.Printf("Parent: %s", action.Label)
+				//}
 				// Find the result of the nodes?
 				continue
 			}
@@ -747,7 +776,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 			if err != nil {
 				log.Printf("Failed deploying %s from image %s: %s", identifier, image, err)
 				log.Printf("Should send status and exit the entire thing?")
-				shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID)
+				//shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID)
 			}
 
 			log.Printf("Adding visited (3): %s", action.Label)
@@ -757,7 +786,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
 
 			// If children of action.ID are NOT in executed:
 			// Remove them from visited.
-			log.Printf("EXECUTED: %#v", executed)
+			//log.Printf("EXECUTED: %#v", executed)
 		}
 
 		//log.Println(nextAction)