From 918d149165a3e3b5e591c7bebc1d24a5a0138e4b Mon Sep 17 00:00:00 2001 From: frikky Date: Mon, 14 Jun 2021 09:08:51 +0200 Subject: [PATCH] #401: Fixed execution issues and added node skipping --- backend/go-app/go.mod | 4 +- backend/go-app/walkoff.go | 4 ++ docker-compose.yml | 2 +- frontend/src/views/Workflows.jsx | 4 +- functions/onprem/worker/build.sh | 2 +- functions/onprem/worker/go.mod | 3 +- functions/onprem/worker/worker.go | 89 ++++++++++++++++++++++++++++--- 7 files changed, 94 insertions(+), 14 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index 7439deaf..47a4753f 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -2,7 +2,7 @@ module shuffle go 1.13 -replace github.com/frikky/shuffle-shared => ../../../../git/shuffle-shared +//replace github.com/frikky/shuffle-shared => ../../../../git/shuffle-shared //replace github.com/frikky/kin-openapi => ../../../../git/kin-openapi @@ -23,7 +23,7 @@ require ( github.com/elastic/go-elasticsearch v0.0.0 // indirect github.com/elastic/go-elasticsearch/v7 v7.12.0 // indirect github.com/frikky/kin-openapi v0.39.0 - github.com/frikky/shuffle-shared v0.0.62 + github.com/frikky/shuffle-shared v0.0.63 github.com/fsouza/go-dockerclient v1.7.2 // indirect github.com/ghodss/yaml v1.0.0 github.com/go-git/go-billy/v5 v5.0.0 diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index e746d7cc..1febda07 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -786,6 +786,10 @@ func validateNewWorkerExecution(body []byte) error { return errors.New(fmt.Sprintf("Bad length of trigger: %d (probably normal app)", len(execution.Workflow.Triggers))) } + if len(baseExecution.Results) >= len(execution.Results) { + return errors.New(fmt.Sprintf("Can't have less actions in a full execution than what exists: %d (old) vs %d (new)", len(baseExecution.Results), len(execution.Results))) + } + //if baseExecution.Status != "WAITING" && baseExecution.Status != "EXECUTING" { // return errors.New(fmt.Sprintf("Workflow is already finished or failed. Can't update")) //} diff --git a/docker-compose.yml b/docker-compose.yml index 53168f1b..c0100d39 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -16,7 +16,7 @@ services: depends_on: - backend backend: - build: ./backend + #build: ./backend image: ghcr.io/frikky/shuffle-backend:nightly container_name: shuffle-backend hostname: ${BACKEND_HOSTNAME} diff --git a/frontend/src/views/Workflows.jsx b/frontend/src/views/Workflows.jsx index d8dae58c..dd865592 100644 --- a/frontend/src/views/Workflows.jsx +++ b/frontend/src/views/Workflows.jsx @@ -86,7 +86,7 @@ export const GetIconInfo = (action) => { const iconList = [ {"key": "cache_add", "values": ["set_cache"]}, {"key": "cache_get", "values": ["get_cache"]}, - {"key": "filter", "values": ["filter"]}, + {"key": "filter", "values": ["filter", "route", "router",]}, {"key": "merge", "values": ["join", "merge"]}, {"key": "search", "values": ["search", "find", "locate", "index",]}, {"key": "list", "values": ["list", "head", "options"]}, @@ -94,7 +94,7 @@ export const GetIconInfo = (action) => { {"key": "add", "values": ["add"]}, {"key": "delete", "values": ["delete", "remove", "clear", "clean",]}, {"key": "send", "values": ["send", "dispatch", "mail", "forward", "post", "submit", "mark", "set"]}, - {"key": "repeat", "values": ["repeat", "retry", "pause",]}, + {"key": "repeat", "values": ["repeat", "retry", "pause", "skip",]}, {"key": "execute", "values": ["execute", "run", "play", "raise",]}, {"key": "extract", "values": ["extract", "unpack", "decompress", "open"]}, {"key": "inflate", "values": ["inflate", "pack", "compress",]}, diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index c5fc2dbf..1910dac5 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-worker -VERSION=0.8.99 +VERSION=0.8.100 echo "Running docker build with $NAME:$VERSION" #CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin . diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index 31879e60..eb62aca7 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -12,8 +12,7 @@ require ( github.com/docker/docker v20.10.5+incompatible github.com/docker/go-connections v0.4.0 // indirect github.com/docker/go-units v0.4.0 // indirect - github.com/elastic/go-elasticsearch/v8 v8.0.0-20210531084204-f01628963386 // indirect - github.com/frikky/shuffle-shared v0.0.59 + github.com/frikky/shuffle-shared v0.0.63 github.com/fsouza/go-dockerclient v1.7.2 github.com/go-git/go-billy/v5 v5.3.1 // indirect github.com/gogo/protobuf v1.3.2 // indirect diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 26286615..1a96e763 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -703,7 +703,18 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { continue } - if action.AppName == "Shuffle Workflow" { + if action.AppName == "Shuffle Tools" && (action.Name == "skip_me" || action.Name == "router" || action.Name == "route") { + err := runSkipAction(topClient, action, workflowExecution.Workflow.ID, workflowExecution.ExecutionId, workflowExecution.Authorization, "SKIPPED") + if err != nil { + log.Printf("[DEBUG] Error in skipme for %s: %s", action.Label, err) + } else { + log.Printf("[INFO] Adding visited (4): %s\n", action.Label) + + visited = append(visited, action.ID) + executed = append(executed, action.ID) + continue + } + } else if action.AppName == "Shuffle Workflow" { //log.Printf("SHUFFLE WORKFLOW: %#v", action) action.Environment = environment action.AppName = "shuffle-subflow" @@ -1275,7 +1286,7 @@ func handleDefaultExecution(client *http.Client, req *http.Request, workflowExec shutdown(workflowExecution, "", "", true) } - log.Printf("DEFAULT EXECUTION Startaction: %s", startAction) + log.Printf("[DEBUG] DEFAULT EXECUTION Startaction: %s", startAction) ctx := context.Background() setWorkflowExecution(ctx, workflowExecution, false) @@ -1388,6 +1399,51 @@ func getAction(workflowExecution shuffle.WorkflowExecution, id, environment stri return shuffle.Action{} } +func runSkipAction(client *http.Client, action shuffle.Action, workflowId, workflowExecutionId, authorization string, configuration string) error { + timeNow := time.Now().Unix() + result := shuffle.ActionResult{ + Action: action, + ExecutionId: workflowExecutionId, + Authorization: authorization, + Result: configuration, + StartedAt: timeNow, + CompletedAt: 0, + Status: "SUCCESS", + } + + resultData, err := json.Marshal(result) + if err != nil { + return err + } + + streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl) + req, err := http.NewRequest( + "POST", + streamUrl, + bytes.NewBuffer([]byte(resultData)), + ) + + if err != nil { + log.Printf("Error building test request: %s", err) + return err + } + + newresp, err := client.Do(req) + if err != nil { + log.Printf("Error running test request: %s", err) + return err + } + + body, err := ioutil.ReadAll(newresp.Body) + if err != nil { + log.Printf("Failed reading body when waiting: %s", err) + return err + } + + log.Printf("[INFO] User Input Body: %s", string(body)) + return nil +} + func runUserInput(client *http.Client, action shuffle.Action, workflowId, workflowExecutionId, authorization string, configuration string) error { timeNow := time.Now().Unix() result := shuffle.ActionResult{ @@ -2051,9 +2107,30 @@ func main() { } } - log.Printf("Environments: %s. Source: %s. 1 = webserver, 0 or >1 = default", environments, workflowExecution.ExecutionSource) - if len(environments) == 1 && workflowExecution.ExecutionSource != "default" { - log.Printf("[INFO] Running OPTIMIZED execution (not manual)") + // Checks if a subflow is child of the startnode, as sub-subflows aren't working properly yet + childNodes := shuffle.FindChildNodes(workflowExecution, workflowExecution.Start) + log.Printf("[DEBUG] Looking for subflow in %#v to check execution pattern as child of %s", childNodes, workflowExecution.Start) + subflowFound := false + for _, childNode := range childNodes { + for _, trigger := range workflowExecution.Workflow.Triggers { + if trigger.ID != childNode { + continue + } + + if trigger.AppName == "Shuffle Workflow" { + subflowFound = true + break + } + } + + if subflowFound { + break + } + } + + log.Printf("\n\nEnvironments: %s. Source: %s. 1 env = webserver, 0 or >1 = default. Subflow exists: %#v\n\n", environments, workflowExecution.ExecutionSource, subflowFound) + if len(environments) == 1 && workflowExecution.ExecutionSource != "default" && !subflowFound { + log.Printf("\n\n[INFO] Running OPTIMIZED execution (not manual)\n\n") listener := webserverSetup(workflowExecution) err := executionInit(workflowExecution) if err != nil { @@ -2073,7 +2150,7 @@ func main() { //wg.Add(1) //wg.Wait() } else { - log.Printf("[INFO] Running NON-OPTIMIZED execution for type %s with %d environments. This only happens when ran manually. Status: %s", workflowExecution.ExecutionSource, len(environments), workflowExecution.Status) + log.Printf("\n\n[INFO] Running NON-OPTIMIZED execution for type %s with %d environments. This only happens when ran manually. Status: %s\n\n", workflowExecution.ExecutionSource, len(environments), workflowExecution.Status) //err := executionInit(workflowExecution) //if err != nil { // log.Printf("[INFO] Workflow setup failed: %s", workflowExecution.ExecutionId, err)