From f4e0446bc91cfca78078e648acf8fb0aa5baf48e Mon Sep 17 00:00:00 2001 From: frikky Date: Fri, 19 Feb 2021 20:07:04 +0100 Subject: [PATCH] #223: Added subflow in same workflow --- backend/app_sdk/app_base.py | 10 ++-- backend/go-app/walkoff.go | 53 +++++++++++++++++--- frontend/src/views/AngularWorkflow.jsx | 69 ++++++++++++++++++++------ functions/onprem/worker/worker.go | 69 +++++++++++++------------- 4 files changed, 138 insertions(+), 63 deletions(-) diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index 3c39976a..8c43637d 100644 --- a/backend/app_sdk/app_base.py +++ b/backend/app_sdk/app_base.py @@ -298,7 +298,7 @@ class AppBase: newparams[key] = value[0] has_loop = True else: - print("Key %s is NOT a list within a list: %s" % (key, value)) + print("Key %s is NOT a list within a list" % (key)) newparams[key] = value @@ -1054,7 +1054,7 @@ class AppBase: except KeyError as error: print(f"KeyError in JSON: {error}") - print(f"[INFO] After first trycatch. Baseresult: ", baseresult) + print(f"[INFO] After first trycatch. Baseresult")#, baseresult) # 2. Find the JSON data if len(baseresult) == 0: @@ -1067,7 +1067,7 @@ class AppBase: baseresult = baseresult.replace(" True,", " true,") baseresult = baseresult.replace(" False", " false,") - print("[INFO] After third parser return - Formatted: ", baseresult) + print("[INFO] After third parser return - Formatted")#, baseresult) basejson = {} try: basejson = json.loads(baseresult) @@ -1570,9 +1570,9 @@ class AppBase: multi_parameters[parameter["name"]] = resultarray multi_execution_lists.append(new_replacement) - print("MULTI finished: %s" % json_replacement) + #print("MULTI finished: %s" % json_replacement) else: - print("(2) Pre replacement: %s" % actualitem) + print("(2) Pre replacement. ") #% actualitem) # This is here to handle for loops within variables.. kindof # 1. Find the length of the longest array # 2. Build an array with the base values based on parameter["value"] diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 9efbcd66..7fab3fc6 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -1163,6 +1163,7 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl log.Printf("[WARNING] Actionresult is %s for node %s in %s. Continuing anyway because of workflow configuration.", actionResult.Status, actionResult.Action.ID, workflowExecution.ExecutionId) // Finds ALL childnodes to set them to SKIPPED childNodes = findChildNodes(*workflowExecution, actionResult.Action.ID) + // Remove duplicates //log.Printf("CHILD NODES: %d", len(childNodes)) for _, nodeId := range childNodes { @@ -1223,6 +1224,7 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl Name: curAction.Name, ID: curAction.ID, } + newResult := ActionResult{ Action: newAction, ExecutionId: actionResult.ExecutionId, @@ -3127,6 +3129,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf } makeNew := true + start, startok := request.URL.Query()["start"] if request.Method == "POST" { body, err := ioutil.ReadAll(request.Body) if err != nil { @@ -3158,7 +3161,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf sourceExecution, sourceExecutionOk := request.URL.Query()["source_execution"] if sourceExecutionOk { - log.Printf("Got source execution%s", sourceExecution) + //log.Printf("[INFO] Got source execution%s", sourceExecution) workflowExecution.ExecutionParent = sourceExecution[0] } else { //log.Printf("Did NOT get source execution") @@ -3226,12 +3229,11 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf // Check for parameters of start and ExecutionId // This is mostly used for user input trigger - start, startok := request.URL.Query()["start"] answer, answerok := request.URL.Query()["answer"] referenceId, referenceok := request.URL.Query()["reference_execution"] if answerok && referenceok { // If answer is false, reference execution with result - log.Printf("Answer is OK AND reference is OK!") + log.Printf("[INFO] Answer is OK AND reference is OK!") if answer[0] == "false" { log.Printf("Should update reference and return, no need for further execution!") @@ -3302,12 +3304,12 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf } // Don't override workflow defaults - if startok { - log.Printf("Setting start to %s based on query!", start[0]) - //workflowExecution.Workflow.Start = start[0] - workflowExecution.Start = start[0] - } + } + if startok { + //log.Printf("\n\n[INFO] Setting start to %s based on query!\n\n", start[0]) + //workflowExecution.Workflow.Start = start[0] + workflowExecution.Start = start[0] } // FIXME - regex uuid, and check if already exists? @@ -3495,7 +3497,42 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf break } } + + if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" { + found := false + for _, node := range childNodes { + if node == trigger.ID { + found = true + break + } + } + + if !found { + log.Printf("SHOULD SET TRIGGER %s TO BE SKIPPED", trigger.ID) + + curaction := Action{ + AppName: trigger.AppName, + AppVersion: trigger.AppVersion, + Label: trigger.Label, + Name: trigger.Name, + ID: trigger.ID, + } + + defaultResults = append(defaultResults, ActionResult{ + Action: curaction, + ExecutionId: workflowExecution.ExecutionId, + Authorization: workflowExecution.Authorization, + Result: "Skipped because it's not under the startnode", + StartedAt: 0, + CompletedAt: 0, + Status: "SKIPPED", + }) + } else { + log.Printf("SHOULD KEEP TRIGGER %s", trigger.ID) + } + } } + //childNodes := findChildNodes(workflowExecution, workflowExecution.Start) if !startFound { log.Printf("Startnode %s doesn't exist!", workflowExecution.Start) diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 59cbd2ca..2fec90dd 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -259,6 +259,7 @@ const AngularWorkflow = (props) => { setWorkflows(responseJson) if (trigger_index > -1) { + var outersub = {} const trigger = workflow.triggers[trigger_index] if (trigger.parameters.length >= 3) { for (var key in trigger.parameters) { @@ -267,8 +268,23 @@ const AngularWorkflow = (props) => { const sub = responseJson.find(data => data.id === param.value) if (sub !== undefined && subworkflow.id !== sub.id) { setSubworkflow(sub) + outersub = sub } } + + if (param.name === "startnode" && outersub.id !== undefined) { + console.log("SHOULD SET STARTNODE: ", outersub) + const innernode = outersub.actions.find(action => action.id === param.value) + console.log("FOUND NODE: ", innernode) + if (innernode !== undefined && subworkflowStartnode.id !== innernode.id) { + setSubworkflowStartnode(innernode) + } + /* + const sub = responseJson.find(data => data.id === param.value) + setSubworkflow(sub) + } + */ + } } } } @@ -391,7 +407,6 @@ const AngularWorkflow = (props) => { //console.log("SHOW EXECUTION ", tmpView) const execution = responseJson.find(data => data.execution_id === tmpView) if (execution !== null && execution !== undefined) { - console.log("EXEC: ", execution) setExecutionData(execution) setExecutionModalView(1) } @@ -5098,6 +5113,7 @@ const AngularWorkflow = (props) => { workflow.triggers[selectedTriggerIndex].parameters[0] = {"name": "workflow", "value": ""} workflow.triggers[selectedTriggerIndex].parameters[1] = {"name": "argument", "value": ""} workflow.triggers[selectedTriggerIndex].parameters[2] = {"name": "user_apikey", "value": ""} + workflow.triggers[selectedTriggerIndex].parameters[3] = {"name": "startnode", "value": ""} console.log("SETTINGS: ", userSettings) if (userSettings !== undefined && userSettings !== null && userSettings.apikey !== null && userSettings.apikey !== undefined && userSettings.apikey.length > 0) { @@ -5159,22 +5175,41 @@ const AngularWorkflow = (props) => { workflow.triggers[selectedTriggerIndex].parameters[0].value = e.target.value.id setWorkflow(workflow) setSubworkflowStartnode(e.target.value.start) + + // Sets the startnode + if (e.target.value.id !== workflow.id) { + const startnode = e.target.value.actions.find(action => action.id === e.target.value.start) + if (startnode !== undefined && startnode !== null) { + setSubworkflowStartnode(startnode) + } + console.log("STARTNODE: ", startnode) + } }} style={{backgroundColor: inputColor, color: "white", height: "50px"}} > {workflows.map((data, index) => { + /* if (data.id === workflow.id) { return null } + */ return ( - + {data.name} ) })} } + {workflow.triggers[selectedTriggerIndex].parameters[0].value.length === 0 ? null : Explore selected workflow} + +
+
+
+ Select the Startnode +
+
{subworkflow === undefined || subworkflow === null || subworkflow.id === undefined || subworkflow.actions === null || subworkflow.actions === undefined || subworkflow.actions.length === 0 ? null : } - {workflow.triggers[selectedTriggerIndex].parameters[0].value.length === 0 ? null : Explore selected workflow}
diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 123b2631..e09ca874 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -1114,16 +1114,16 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { if isSkipped { //log.Printf("Skipping %s as all parents are done", item.Action.Label) if !arrayContains(visited, item.Action.ID) { - log.Printf("Adding visited (1): %s", item.Action.Label) + log.Printf("[INFO] Adding visited (1): %s", item.Action.Label) visited = append(visited, item.Action.ID) } } else { - log.Printf("Continuing %s as all parents are NOT done", item.Action.Label) + log.Printf("[INFO] Continuing %s as all parents are NOT done", item.Action.Label) appendActions = append(appendActions, item.Action.ID) } } else { if item.Status == "FINISHED" { - log.Printf("Adding visited (2): %s", item.Action.Label) + log.Printf("[INFO] Adding visited (2): %s", item.Action.Label) visited = append(visited, item.Action.ID) } } @@ -1149,7 +1149,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { // care if it gets stuck in a loop. // FIXME: Force killing a worker should result in a notification somewhere if len(nextActions) == 0 { - log.Printf("No next action. Finished? Result vs Actions: %d - %d", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) + log.Printf("[INFO] No next action. Finished? Result vs Actions: %d - %d", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) exit := true for _, item := range workflowExecution.Results { if item.Status == "EXECUTING" { @@ -1235,6 +1235,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { // IF NOT VISITED && IN toExecuteOnPrem // SKIP if it's not onprem toRemove := []int{} + log.Printf("\n\nNEXTACTIONS: %#v\n\n", nextActions) for index, nextAction := range nextActions { action := getAction(workflowExecution, nextAction, environment) // check visited and onprem @@ -1377,7 +1378,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { } if continueOuter { - log.Printf("Parents of %s aren't finished: %s", nextAction, strings.Join(parents[nextAction], ", ")) + log.Printf("[INFO] Parents of %s aren't finished: %s", nextAction, strings.Join(parents[nextAction], ", ")) //for _, tmpaction := range parents[nextAction] { // action := getAction(workflowExecution, tmpaction) // _ = action @@ -1390,10 +1391,10 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { // get action status actionResult := getResult(workflowExecution, nextAction) if actionResult.Action.ID == action.ID { - log.Printf("%s already has status %s.", action.ID, actionResult.Status) + log.Printf("[INFO] %s already has status %s.", action.ID, actionResult.Status) continue } else { - log.Printf("%s:%s has no status result yet. Should execute.", action.Name, action.ID) + log.Printf("[INFO] %s:%s has no status result yet. Should execute.", action.Name, action.ID) } appname := action.AppName @@ -1445,7 +1446,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { } // marshal action and put it in there rofl - log.Printf("Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters)) + log.Printf("[INFO] Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters)) actionData, err := json.Marshal(action) if err != nil { @@ -1468,7 +1469,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { // Sending full execution so that it won't have to load in every app // This might be an issue if they can read environments, but that's alright // if everything is generated during execution - log.Printf("Deployed with CALLBACK_URL %s and BASE_URL %s", appCallbackUrl, baseUrl) + log.Printf("[INFO] Deployed with CALLBACK_URL %s and BASE_URL %s", appCallbackUrl, baseUrl) env := []string{ fmt.Sprintf("ACTION=%s", string(actionData)), fmt.Sprintf("EXECUTIONID=%s", workflowExecution.ExecutionId), @@ -1593,7 +1594,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { } } - log.Printf("Adding visited (3): %s", action.Label) + log.Printf("[INFO] Adding visited (3): %s", action.Label) visited = append(visited, action.ID) executed = append(executed, action.ID) @@ -1636,14 +1637,22 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { func executionInit(workflowExecution WorkflowExecution) error { parents = map[string][]string{} children = map[string][]string{} - triggersHandled := []string{} startAction = workflowExecution.Start + log.Printf("[INFO] STARTACTION: %s", startAction) if len(startAction) == 0 { - log.Printf("Didn't find execution start action. Setting it to workflow start action.") + log.Printf("[INFO] Didn't find execution start action. Setting it to workflow start action.") startAction = workflowExecution.Workflow.Start } + // Setting up extra counter + for _, trigger := range workflowExecution.Workflow.Triggers { + //log.Printf("Appname trigger (0): %s", trigger.AppName) + if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" { + extra += 1 + } + } + nextActions = append(nextActions, startAction) for _, branch := range workflowExecution.Workflow.Branches { // Check what the parent is first. If it's trigger - skip @@ -1662,27 +1671,15 @@ func executionInit(workflowExecution WorkflowExecution) error { for _, trigger := range workflowExecution.Workflow.Triggers { //log.Printf("Appname trigger (0): %s", trigger.AppName) if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" { - //log.Printf("%s is a special trigger. Checking where.", trigger.AppName) - - found := false - for _, check := range triggersHandled { - if check == trigger.ID { - found = true - break - } - } - - if !found { - extra += 1 - } else { - triggersHandled = append(triggersHandled, trigger.ID) + if branch.SourceID == "c9560766-3f85-4589-8324-311acd6be820" { + log.Printf("BRANCH: %#v", branch) } if trigger.ID == branch.SourceID { - log.Printf("Trigger %s is the source!", trigger.AppName) + log.Printf("[INFO] Trigger %s is the source!", trigger.AppName) sourceFound = true } else if trigger.ID == branch.DestinationID { - log.Printf("Trigger %s is the destination!", trigger.AppName) + log.Printf("[INFO] Trigger %s is the destination!", trigger.AppName) destinationFound = true } } @@ -1691,17 +1688,21 @@ func executionInit(workflowExecution WorkflowExecution) error { 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) + log.Printf("[INFO] 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("[INFO] ID %s was not found in actions! Skipping child. (TRIGGER?)", branch.SourceID) } } - log.Printf("Actions: %d + Special Triggers: %d", len(workflowExecution.Workflow.Actions), extra) + log.Printf("\n\n\n[INFO] CHILDREN FOUND: %#v", children) + log.Printf("[INFO] PARENTS FOUND: %#v", parents) + log.Printf("[INFO] NEXT ACTIONS: %#v\n\n", nextActions) + + log.Printf("[INFO] Actions: %d + Special Triggers: %d", len(workflowExecution.Workflow.Actions), extra) onpremApps := []string{} toExecuteOnprem := []string{} for _, action := range workflowExecution.Workflow.Actions { @@ -1730,7 +1731,7 @@ func executionInit(workflowExecution WorkflowExecution) error { pullOptions := types.ImagePullOptions{} _ = pullOptions for _, image := range onpremApps { - log.Printf("Image: %s", image) + log.Printf("[INFO] Image: %s", image) // Kind of gambling that the image exists. if strings.Contains(image, " ") { image = strings.ReplaceAll(image, " ", "-") @@ -2498,7 +2499,7 @@ func getWorkflowExecution(ctx context.Context, id string) (*WorkflowExecution, e } func validateFinished(workflowExecution WorkflowExecution) { - log.Printf("Status: %s, Actions: %d, Extra: %d, Results: %d\n", workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results)) + log.Printf("[INFO] Status: %s, Actions: %d, Extra: %d, Results: %d\n", workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results)) //if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions)+extra { if (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1) || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions) && len(workflowExecution.Workflow.Actions) > 0) { @@ -2703,7 +2704,7 @@ func main() { } else { authorization = os.Getenv("AUTHORIZATION") executionId = os.Getenv("EXECUTIONID") - log.Printf("Running normal execution with auth %s and ID %s", authorization, executionId) + log.Printf("[INFO] Running normal execution with auth %s and ID %s", authorization, executionId) } if len(authorization) == 0 {