From 3c6e973f3d80551aac4cbce7fc9d55e4a8bc33c3 Mon Sep 17 00:00:00 2001 From: frikky Date: Tue, 2 Mar 2021 08:58:32 +0100 Subject: [PATCH] BUGFIX: Subflows and workflow abort issues --- backend/go-app/main.go | 12 +- backend/go-app/walkoff.go | 82 +++++++++++++- frontend/src/defaultCytoscapeStyle.js | 2 +- frontend/src/views/AngularWorkflow.jsx | 28 +++-- functions/onprem/orborus/orborus.go | 8 +- functions/onprem/worker/build.sh | 2 +- functions/onprem/worker/worker.go | 148 ++++++++++++++++--------- 7 files changed, 205 insertions(+), 77 deletions(-) diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 7cd4e069..d1d5c5aa 100644 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -3559,9 +3559,8 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) { // bodyWrapper = string(parsedBody) //} - url := &url.URL{} newRequest := &http.Request{ - URL: url, + URL: &url.URL{}, Method: "POST", Body: ioutil.NopCloser(bytes.NewReader(b)), } @@ -7234,9 +7233,10 @@ func runInit(ctx context.Context) { if count == 0 && err == nil && len(activeOrgs) == 1 { log.Printf("Setting up environment with org %s", activeOrgs[0].Id) item := Environment{ - Name: "Shuffle", - Type: "onprem", - OrgId: activeOrgs[0].Id, + Name: "Shuffle", + Type: "onprem", + OrgId: activeOrgs[0].Id, + Default: true, } err = setEnvironment(ctx, &item) @@ -7419,6 +7419,7 @@ func runInit(ctx context.Context) { log.Printf("Failed getting schedules during service init: %s", err) } else { log.Printf("Setting up %d schedule(s)", len(schedules)) + url := &url.URL{} for _, schedule := range schedules { if schedule.Environment == "cloud" { log.Printf("Skipping cloud schedule") @@ -7428,6 +7429,7 @@ func runInit(ctx context.Context) { //log.Printf("Schedule: %#v", schedule) job := func() { request := &http.Request{ + URL: url, Method: "POST", Body: ioutil.NopCloser(strings.NewReader(schedule.WrappedArgument)), } diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 9c0556db..0588c6bd 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -10,6 +10,7 @@ import ( "io/ioutil" "log" "net/http" + "net/url" "os" "sort" "strconv" @@ -650,6 +651,7 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode log.Printf("WRAPPER BODY: \n%s", bodyWrapper) job := func() { request := &http.Request{ + URL: &url.URL{}, Method: "POST", Body: ioutil.NopCloser(strings.NewReader(bodyWrapper)), } @@ -1014,7 +1016,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Success"}`))) return } else { - //log.Printf("[WARNING] Failed to handle new execution variant: %s", err) + //log.Printf("[WARNING] Handling other execution variant: %s", err) } var actionResult ActionResult @@ -2899,6 +2901,8 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) { return } + //log.Printf("\n\nINSIDE ABORT\n\n") + location := strings.Split(request.URL.String(), "/") var fileId string if location[1] == "api" { @@ -2974,6 +2978,7 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) { workflowExecution.CompletedAt = int64(time.Now().Unix()) workflowExecution.Status = "ABORTED" + log.Printf("[INFO] Running shutdown of %s", workflowExecution.ExecutionId) lastResult := "" newResults := []ActionResult{} @@ -2996,6 +3001,79 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) { workflowExecution.Result = lastResult } + addResult := true + for _, result := range workflowExecution.Results { + if result.Status != "SKIPPED" { + addResult = false + } + } + + extra := 0 + 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 + } + } + + parsedReason := "An error occurred during execution of this node" + reason, reasonok := request.URL.Query()["reason"] + if reasonok { + parsedReason = reason[0] + } + + if len(workflowExecution.Results) == 0 || addResult { + newaction := Action{ + ID: workflowExecution.Start, + } + + for _, action := range workflowExecution.Workflow.Actions { + if action.ID == workflowExecution.Start { + newaction = action + break + } + } + + workflowExecution.Results = append(workflowExecution.Results, ActionResult{ + Action: newaction, + ExecutionId: workflowExecution.ExecutionId, + Authorization: workflowExecution.Authorization, + Result: parsedReason, + StartedAt: workflowExecution.StartedAt, + CompletedAt: workflowExecution.StartedAt, + Status: "FAILURE", + }) + } else if len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions)+extra { + log.Printf("[INFO] DONE - Nothing to add during abort!") + } else { + //log.Printf("VALIDATING INPUT!") + node, nodeok := request.URL.Query()["node"] + if nodeok { + nodeId := node[0] + log.Printf("[INFO] Found abort node %s", nodeId) + newaction := Action{ + ID: nodeId, + } + + for _, action := range workflowExecution.Workflow.Actions { + if action.ID == nodeId { + newaction = action + break + } + } + + workflowExecution.Results = append(workflowExecution.Results, ActionResult{ + Action: newaction, + ExecutionId: workflowExecution.ExecutionId, + Authorization: workflowExecution.Authorization, + Result: parsedReason, + StartedAt: workflowExecution.StartedAt, + CompletedAt: workflowExecution.StartedAt, + Status: "FAILURE", + }) + } + } + err = setWorkflowExecution(ctx, *workflowExecution, true) if err != nil { log.Printf("Error saving workflow execution for updates when aborting %s: %s", topic, err) @@ -3517,7 +3595,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf //log.Printf("SHOULD SET TRIGGER %s TO BE SKIPPED", trigger.ID) curaction := Action{ - AppName: trigger.AppName, + AppName: "shuffle-subflow", AppVersion: trigger.AppVersion, Label: trigger.Label, Name: trigger.Name, diff --git a/frontend/src/defaultCytoscapeStyle.js b/frontend/src/defaultCytoscapeStyle.js index e0732bac..ab8fae1e 100644 --- a/frontend/src/defaultCytoscapeStyle.js +++ b/frontend/src/defaultCytoscapeStyle.js @@ -183,7 +183,7 @@ const data = [{ css: { 'background-color': "#f85a3e", 'border-color': '#f85a3e', - 'border-width': '5px', + 'border-width': '8px', 'transition-property': 'border-width', 'transition-duration': '0.25s', }, diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 4cd2a35b..0fc7178b 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -786,7 +786,7 @@ const AngularWorkflow = (props) => { curelements[i].addClass("not-executing-highlight") } - if (executionArgument.length > 0) { + if (executionArgument !== undefined && executionArgument !== null && executionArgument.length > 0) { //alert.success("Starting execution WITH an execution argument") } else { //alert.success("Starting execution") @@ -2794,8 +2794,10 @@ const AngularWorkflow = (props) => { } var exampledata = item.example === undefined ? "" : item.example + console.log("EXAMPLE: ", exampledata) // Find previous execution and their variables - if (exampledata === "" && workflowExecutions.length > 0) { + //exampledata === "" && + if (workflowExecutions.length > 0) { // Look for the ID const found = false for (var key in workflowExecutions) { @@ -3172,8 +3174,8 @@ const AngularWorkflow = (props) => { { setMenuPosition({ - top: event.pageY, - left: event.pageX, + top: event.pageY+10, + left: event.pageX+10, }) setShowDropdownNumber(count) setShowDropdown(true) @@ -3235,8 +3237,8 @@ const AngularWorkflow = (props) => { { setMenuPosition({ - top: event.pageY, - left: event.pageX, + top: event.pageY+10, + left: event.pageX+10, }) setShowDropdownNumber(count) setShowDropdown(true) @@ -3586,7 +3588,7 @@ const AngularWorkflow = (props) => { // FIXME: Should be recursive in here const icon = pathdata.type === "value" ? : pathdata.type === "list" ? : return ( - {}} + {}} onClick={() => { handleItemClick([innerdata, pathdata]) }} @@ -6663,19 +6665,25 @@ const AngularWorkflow = (props) => { if (action !== undefined && action !== null) { imgSrc = action.large_image } + + /* + if (imgSrc.length === 0) { + console.log("CHECK IF ITS A + } + */ } var actionimg = curapp === null ? null : - {data.action.app_name} + {data.action.app_name} if (triggers.length > 2) { if (data.action.app_name === "shuffle-subflow") { - actionimg = {"Shuffle + actionimg = {"Shuffle } if (data.action.app_name === "User Input") { - actionimg = {"Shuffle + actionimg = {"Shuffle } } diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index a2e3d716..30d6394a 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -101,7 +101,7 @@ func getThisContainerId() { log.Printf("[INFO] Running containerized in Docker!") default: - fCol = "3" // for backward-compatibility with production + fCol = "0" // for backward-compatibility with production log.Printf("[WARNING] RUNNING_MODE not set - defaulting to Docker (NOT Kubernetes).") } @@ -119,8 +119,10 @@ func getThisContainerId() { //docker-76c537e9a4b7c7233011f5d70e6b7f2d600b6413ac58a96519b8dca7a3f7117a.scope } } else { - containerId = "shuffle-orborus" - log.Printf("[WARNING] Failed getting container ID: %s", err) + if fCol != "0" { + containerId = "shuffle-orborus" + log.Printf("[WARNING] Failed getting container ID: %s", err) + } } } diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index ba1b2e38..907f053e 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -10,5 +10,5 @@ docker build . -t frikky/shuffle:$NAME -t frikky/shuffle:$NAME_$VERSION -t docke #docker push frikky/shuffle:$NAME_$VERSION #docker push docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION #docker tag frikky/shuffle:0.8.51 ghcr.io/frikky/shuffle-worker:0.8.5 -docker tag frikky/shuffle:$NAME ghcr.io/frikky/shuffle-worker:0.8.52 +#docker tag frikky/shuffle:$NAME ghcr.io/frikky/shuffle-worker:0.8.52 docker push ghcr.io/frikky/$NAME:$VERSION diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 1672fb22..f3d9250a 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -11,6 +11,7 @@ import ( "log" "net" "net/http" + "net/url" "os" "os/exec" "strings" @@ -774,8 +775,22 @@ type AppExecutionExample struct { } // removes every container except itself (worker) -func shutdown(executionId, workflowId string) { +func shutdown(workflowExecution WorkflowExecution, nodeId string, reason string, handleResultSend bool) { log.Printf("[INFO] Shutdown started") + //reason := "Error in execution" + + sleepDuration := 1 + if handleResultSend { + data, err := json.Marshal(workflowExecution) + if err == nil { + sendResult(workflowExecution, data) + log.Printf("[WARNING] Sent shutdown update") + } else { + log.Printf("[WARNING] DIDNT send update") + } + + time.Sleep(time.Duration(sleepDuration) * time.Second) + } // Might not be necessary because of cleanupEnv hostconfig autoremoval if cleanupEnv == "true" && len(containerIds) > 0 { @@ -801,7 +816,20 @@ func shutdown(executionId, workflowId string) { log.Printf("[INFO] NOT cleaning up containers. IDS: %d, CLEANUP env: %s", len(containerIds), cleanupEnv) } - fullUrl := fmt.Sprintf("%s/api/v1/workflows/%s/executions/%s/abort", baseUrl, workflowId, executionId) + fullUrl := fmt.Sprintf("%s/api/v1/workflows/%s/executions/%s/abort", baseUrl, workflowExecution.Workflow.ID, workflowExecution.ExecutionId) + + path := fmt.Sprintf("?reason=%s", url.QueryEscape(reason)) + if len(nodeId) > 0 { + path += fmt.Sprintf("&node=%s", url.QueryEscape(nodeId)) + } + if len(environment) > 0 { + path += fmt.Sprintf("&env=%s", url.QueryEscape(environment)) + } + + //fmt.Println(url.QueryEscape(query)) + fullUrl += path + log.Printf("Abort URL: %s", fullUrl) + req, err := http.NewRequest( "GET", fullUrl, @@ -844,7 +872,6 @@ func shutdown(executionId, workflowId string) { log.Printf("[INFO] Failed abort request: %s", err) } - sleepDuration := 1 log.Printf("[INFO] Finished shutdown (after %d seconds).", sleepDuration) // Allows everything to finish in subprocesses time.Sleep(time.Duration(sleepDuration) * time.Second) @@ -931,7 +958,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] err = cli.ContainerStart(context.Background(), cont.ID, types.ContainerStartOptions{}) if err != nil { log.Printf("[ERROR] Failed to start container in environment %s: %s", environment, err) - //shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + //shutdown(workflowExecution, workflowExecution.Workflow.ID, true) return err } @@ -1202,7 +1229,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { if exit && len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions) { log.Printf("Shutting down.") - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } // Look for the NEXT missing action @@ -1547,18 +1574,18 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { reader, err := dockercli.ImagePull(context.Background(), image, pullOptions) if err != nil { log.Printf("[ERROR] Failed getting %s. The couldn't be find locally, AND is missing.", image) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } buildBuf := new(strings.Builder) _, err = io.Copy(buildBuf, reader) if err != nil { log.Printf("[ERROR] Error in IO copy: %s", err) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } else { if strings.Contains(buildBuf.String(), "errorDetail") { log.Printf("[ERROR] Docker build:\n%s\nERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), image) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } log.Printf("[INFO] Successfully downloaded %s", image) @@ -1571,7 +1598,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { if strings.Contains(err.Error(), "No such image") { //log.Printf("[WARNING] Failed deploying %s from image %s: %s", identifier, image, err) log.Printf("[ERROR] Image doesn't exist. Shutting down") - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } } } @@ -1599,18 +1626,18 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { reader, err := dockercli.ImagePull(context.Background(), image, pullOptions) if err != nil { log.Printf("[ERROR] Failed getting %s. The couldn't be find locally, AND is missing.", image) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } buildBuf := new(strings.Builder) _, err = io.Copy(buildBuf, reader) if err != nil { log.Printf("[ERROR] Error in IO copy: %s", err) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } else { if strings.Contains(buildBuf.String(), "errorDetail") { log.Printf("[ERROR] Docker build:\n%s\nERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), image) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } log.Printf("[INFO] Successfully downloaded %s", image) @@ -1623,7 +1650,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { if strings.Contains(err.Error(), "No such image") { //log.Printf("[WARNING] Failed deploying %s from image %s: %s", identifier, image, err) log.Printf("[ERROR] Image doesn't exist. Shutting down") - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, action.ID, err.Error(), true) } } } @@ -1663,7 +1690,7 @@ func handleExecutionResult(workflowExecution WorkflowExecution) { if shutdownCheck { log.Println("[INFO] BREAKING BECAUSE RESULTS IS SAME LENGTH AS ACTIONS. SHOULD CHECK ALL RESULTS FOR WHETHER THEY'RE DONE") validateFinished(workflowExecution) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } } @@ -1781,7 +1808,7 @@ func executionInit(workflowExecution WorkflowExecution) error { //reader, err := dockercli.ImagePull(context.Background(), image, pullOptions) //if err != nil { // log.Printf("Failed getting %s. The app is missing or some other issue", image) - // shutdown(workflowExecution.ExecutionId) + // shutdown(workflowExecution) //} ////io.Copy(os.Stdout, reader) @@ -1799,7 +1826,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W err := executionInit(workflowExecution) if err != nil { log.Printf("[INFO] Workflow setup failed: %s", workflowExecution.ExecutionId, err) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } log.Printf("Startaction: %s", startAction) @@ -1835,7 +1862,11 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W if newresp.StatusCode != 200 { log.Printf("[ERROR] Bad statuscode: %d, %s", newresp.StatusCode, string(body)) - //shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + + if strings.Contains(string(body), "Workflowexecution is already finished") { + shutdown(workflowExecution, "", "", false) + } + time.Sleep(time.Duration(sleepTime) * time.Second) continue } @@ -1849,13 +1880,13 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" { log.Printf("[INFO] Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } log.Printf("[INFO] Status: %s, Results: %d, actions: %d", workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra) if workflowExecution.Status != "EXECUTING" { log.Printf("[WARNING] Exiting as worker execution has status %s!", workflowExecution.Status) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } } @@ -2545,6 +2576,34 @@ func getWorkflowExecution(ctx context.Context, id string) (*WorkflowExecution, e return &WorkflowExecution{}, errors.New("No workflowexecution defined yet") } +func sendResult(workflowExecution WorkflowExecution, data []byte) { + fullUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl) + req, err := http.NewRequest( + "POST", + fullUrl, + bytes.NewBuffer([]byte(data)), + ) + + if err != nil { + log.Printf("[ERROR] Failed creating finishing request: %s", err) + shutdown(workflowExecution, "", "", false) + } + + newresp, err := topClient.Do(req) + if err != nil { + log.Printf("[ERROR] Error running finishing request: %s", err) + shutdown(workflowExecution, "", "", false) + } + + body, err := ioutil.ReadAll(newresp.Body) + log.Printf("[INFO] BACKEND STATUS: %d", newresp.StatusCode) + if err != nil { + log.Printf("[ERROR] Failed reading body: %s", err) + } else { + log.Printf("[INFO] NEWRESP (from backend): %s", string(body)) + } +} + func validateFinished(workflowExecution WorkflowExecution) { log.Printf("[INFO] Status: %s, Actions: %d, Extra: %d, Results: %d\n", workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results)) @@ -2557,34 +2616,10 @@ func validateFinished(workflowExecution WorkflowExecution) { data, err := json.Marshal(workflowExecution) if err != nil { log.Printf("[ERROR] Failed to unmarshal data for backend") - shutdown(workflowExecution.ExecutionId, "") + shutdown(workflowExecution, "", "", true) } - fullUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl) - req, err := http.NewRequest( - "POST", - fullUrl, - bytes.NewBuffer([]byte(data)), - ) - - if err != nil { - log.Printf("[ERROR] Failed creating finishing request: %s", err) - shutdown(workflowExecution.ExecutionId, "") - } - - newresp, err := topClient.Do(req) - if err != nil { - log.Printf("[ERROR] Error running finishing request: %s", err) - shutdown(workflowExecution.ExecutionId, "") - } - - body, err := ioutil.ReadAll(newresp.Body) - log.Printf("[INFO] BACKEND STATUS: %d", newresp.StatusCode) - if err != nil { - log.Printf("[ERROR] Failed reading body: %s", err) - } else { - log.Printf("[INFO] NEWRESP (from backend): %s", string(body)) - } + sendResult(workflowExecution, data) } } @@ -2649,8 +2684,9 @@ func setWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecuti handleExecutionResult(workflowExecution) validateFinished(workflowExecution) if dbSave { - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", false) } + return nil } @@ -2691,7 +2727,7 @@ func webserverSetup(workflowExecution WorkflowExecution) net.Listener { listener, err := getAvailablePort() if err != nil { log.Printf("Failed to created listener: %s", err) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } port := listener.Addr().(*net.TCPAddr).Port @@ -2754,14 +2790,17 @@ func main() { log.Printf("[INFO] Running normal execution with auth %s and ID %s", authorization, executionId) } + workflowExecution := WorkflowExecution{ + ExecutionId: executionId, + } if len(authorization) == 0 { log.Println("[INFO] No AUTHORIZATION key set in env") - shutdown(executionId, "") + shutdown(workflowExecution, "", "", false) } if len(executionId) == 0 { log.Println("[INFO] No EXECUTIONID key set in env") - shutdown(executionId, "") + shutdown(workflowExecution, "", "", false) } data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization) @@ -2774,7 +2813,7 @@ func main() { if err != nil { log.Println("[ERROR] Failed making request builder for backend") - shutdown(executionId, "") + shutdown(workflowExecution, "", "", true) } topClient = client @@ -2802,7 +2841,6 @@ func main() { continue } - var workflowExecution WorkflowExecution err = json.Unmarshal(body, &workflowExecution) if err != nil { log.Printf("[ERROR] Failed workflowExecution unmarshal: %s", err) @@ -2837,7 +2875,7 @@ func main() { err := executionInit(workflowExecution) if err != nil { log.Printf("[INFO] Workflow setup failed: %s", workflowExecution.ExecutionId, err) - shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } go func() { @@ -2856,7 +2894,7 @@ func main() { if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" { log.Printf("[INFO] Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId) - shutdown(executionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } if workflowExecution.Status == "EXECUTING" || workflowExecution.Status == "RUNNING" { @@ -2864,11 +2902,11 @@ func main() { err = handleExecution(client, req, workflowExecution) if err != nil { log.Printf("[INFO] Workflow %s is finished: %s", workflowExecution.ExecutionId, err) - shutdown(executionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, "", "", true) } } else { log.Printf("[INFO] Workflow %s has status %s. Exiting worker.", workflowExecution.ExecutionId, workflowExecution.Status) - shutdown(executionId, workflowExecution.Workflow.ID) + shutdown(workflowExecution, workflowExecution.Workflow.ID, "", true) } time.Sleep(time.Duration(sleepTime) * time.Second)