diff --git a/.env b/.env
index 4ffc4215..ac31f93a 100644
--- a/.env
+++ b/.env
@@ -29,8 +29,9 @@ BACKEND_HOSTNAME=shuffle-backend
BACKEND_PORT=5001
FRONTEND_PORT=3001
FRONTEND_PORT_HTTPS=3443
+# CHANGE THIS IF YOU WANT GOOD LOCAL EXECUTIONS:
OUTER_HOSTNAME=shuffle-backend
-DB_LOCATION=./shuffle-database
+DB_LOCATION=./shuffle-database-new
# Proxy configurations. SHUFFLE_PASS_WORKER_PROXY must be FALSE to not pass the proxy information to sub-apps.
# PS: It will skip proxy for
diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py
index 75d0acb4..29aa07cd 100644
--- a/backend/app_sdk/app_base.py
+++ b/backend/app_sdk/app_base.py
@@ -2371,7 +2371,6 @@ class AppBase:
# self.action = cls
app = cls(redis=None, logger=logger, console_logger=logger)
-
if isinstance(action, str):
print("Normal execution. Action is a string.")
elif isinstance(action, object):
diff --git a/backend/app_sdk/build.sh b/backend/app_sdk/build.sh
index ca82bf90..f760beee 100644
--- a/backend/app_sdk/build.sh
+++ b/backend/app_sdk/build.sh
@@ -1,6 +1,6 @@
#!/bin/bash
NAME=shuffle-app_sdk
-VERSION=0.8.72
+VERSION=0.8.73
docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force
docker build . -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION
diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod
index 58cb8459..c654792d 100644
--- a/backend/go-app/go.mod
+++ b/backend/go-app/go.mod
@@ -20,7 +20,7 @@ require (
github.com/docker/go-connections v0.4.0
github.com/docker/go-units v0.4.0 // indirect
github.com/frikky/kin-openapi v0.38.0
- github.com/frikky/shuffle-shared v0.0.27
+ github.com/frikky/shuffle-shared v0.0.28
github.com/ghodss/yaml v1.0.0
github.com/go-git/go-billy/v5 v5.0.0
github.com/go-git/go-git/v5 v5.0.0
diff --git a/backend/go-app/go.sum b/backend/go-app/go.sum
index 8d8cf6ec..1360ecc7 100644
--- a/backend/go-app/go.sum
+++ b/backend/go-app/go.sum
@@ -100,6 +100,8 @@ github.com/frikky/shuffle-shared v0.0.22 h1:TFMcJCNmOOSneMMWbg5dNzp2z6m0aZLROuL+
github.com/frikky/shuffle-shared v0.0.22/go.mod h1:H7SqOta/EAYnfYuWzwzYSh/oWfF0kgnuaJTQNKQBvoQ=
github.com/frikky/shuffle-shared v0.0.23 h1:Pnlc2M6fHnFRLFd5K1iLTVv4/t4P04Ri1GJ5CMxzq0U=
github.com/frikky/shuffle-shared v0.0.23/go.mod h1:H7SqOta/EAYnfYuWzwzYSh/oWfF0kgnuaJTQNKQBvoQ=
+github.com/frikky/shuffle-shared v0.0.27 h1:BbibbAv3a5GWR/DfaoSC4D9+fh2cwSEvn9H+EVfd7BM=
+github.com/frikky/shuffle-shared v0.0.27/go.mod h1:H7SqOta/EAYnfYuWzwzYSh/oWfF0kgnuaJTQNKQBvoQ=
github.com/getkin/kin-openapi v0.8.0 h1:a6TQjTqwkyscC4/hShJX7WhCVE+4bi9lzw61XHQW5hE=
github.com/getkin/kin-openapi v0.8.0/go.mod h1:zZQMFkVgRHCdhgb6ihCTIo9dyDZFvX0k/xAKqw1FhPw=
github.com/getkin/kin-openapi v0.52.0 h1:6WqsF5d6PfJ8AscdD+9Rtb2RP2iBWyC7V6GcjssWg7M=
diff --git a/backend/go-app/main.go b/backend/go-app/main.go
index dd437a88..6417dd55 100644
--- a/backend/go-app/main.go
+++ b/backend/go-app/main.go
@@ -5023,7 +5023,9 @@ func runInit(ctx context.Context) {
_, err = dbclient.GetAll(ctx, q, &users)
if err == nil {
setOrgBool := false
+ usernames := []string{}
for _, user := range users {
+ usernames = append(usernames, user.Username)
newUser := shuffle.User{
Username: user.Username,
Id: user.Id,
@@ -5048,6 +5050,8 @@ func runInit(ctx context.Context) {
}
}
+ log.Printf("Users found: %s", strings.Join(usernames, ", "))
+
if setOrgBool {
err = shuffle.SetOrg(ctx, activeOrg, activeOrg.Id)
if err != nil {
@@ -5076,9 +5080,11 @@ func runInit(ctx context.Context) {
if err != nil && len(activeusers) == 0 {
log.Printf("Error getting users during init: %s", err)
} else {
+ log.Printf("Parsing all users and setting them to active.")
q := datastore.NewQuery("Users")
var users []shuffle.User
_, err := dbclient.GetAll(ctx, q, &users)
+ //log.Printf("User ret: %s", err)
if len(activeusers) == 0 && len(users) > 0 {
log.Printf("No active users found - setting ALL to active")
@@ -5136,7 +5142,7 @@ func runInit(ctx context.Context) {
}
}
} else {
- if len(users) < 5 && len(users) > 0 {
+ if len(users) < 10 && len(users) > 0 {
for _, user := range users {
log.Printf("[INFO] Username: %s, role: %s", user.Username, user.Role)
}
@@ -5414,7 +5420,7 @@ func runInit(ctx context.Context) {
if workflowapp.Edited == 0 {
err = shuffle.SetWorkflowAppDatastore(ctx, workflowapp, workflowapp.ID)
if err == nil {
- log.Printf("Updating time for workflowapp %s:%s", workflowapp.Name, workflowapp.AppVersion)
+ log.Printf("[INFO] Updating time for workflowapp %s:%s", workflowapp.Name, workflowapp.AppVersion)
}
}
}
@@ -6115,7 +6121,7 @@ func initHandlers() {
// Make user related locations
// Fix user changes with org
- r.HandleFunc("/api/v1/users/login", handleLogin).Methods("POST", "OPTIONS")
+ r.HandleFunc("/api/v1/users/login", shuffle.HandleLogin).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/users/register", handleRegister).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/users/checkusers", checkAdminLogin).Methods("GET", "OPTIONS")
r.HandleFunc("/api/v1/users/getinfo", handleInfo).Methods("GET", "OPTIONS")
diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go
index 4952f40f..83ca6c12 100644
--- a/backend/go-app/walkoff.go
+++ b/backend/go-app/walkoff.go
@@ -881,61 +881,9 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
}
-// Finds the child nodes of a node in execution and returns them
-// Used if e.g. a node in a branch is exited, and all children have to be stopped
-func findChildNodes(workflowExecution shuffle.WorkflowExecution, nodeId string) []string {
- //log.Printf("\nNODE TO FIX: %s\n\n", nodeId)
- allChildren := []string{nodeId}
-
- // 1. Find children of this specific node
- // 2. Find the children of those nodes etc.
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.SourceID == nodeId {
- //log.Printf("Children: %s", branch.DestinationID)
- allChildren = append(allChildren, branch.DestinationID)
-
- childNodes := findChildNodes(workflowExecution, branch.DestinationID)
- for _, bottomChild := range childNodes {
- found := false
- for _, topChild := range allChildren {
- if topChild == bottomChild {
- found = true
- break
- }
- }
-
- if !found {
- allChildren = append(allChildren, bottomChild)
- }
- }
- }
- }
-
- // 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
-}
-
// Checks if data is sent from Worker >0.8.51, which sends a full execution
// instead of individial results
func validateNewWorkerExecution(body []byte) error {
- //type WorkflowExecution struct {
- //}
-
ctx := context.Background()
var execution shuffle.WorkflowExecution
err := json.Unmarshal(body, &execution)
@@ -981,7 +929,7 @@ func validateNewWorkerExecution(body []byte) error {
execution.Status = "FINISHED"
}
- log.Printf("[INFO] BASEEXECUTION LENGTH: %d", len(baseExecution.Workflow.Actions)+extra)
+ //log.Printf("[INFO] BASEEXECUTION LENGTH: %d", len(baseExecution.Workflow.Actions)+extra)
}
// FIXME: Add extra here
@@ -1035,6 +983,8 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
return
}
+ //log.Printf("Received action: %#v", actionResult)
+
// 1. Get the WorkflowExecution(ExecutionId) from the database
// 2. if ActionResult.Authentication != WorkflowExecution.Authentication -> exit
// 3. Add to and update actionResult in workflowExecution
@@ -1144,443 +1094,58 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl
return
}
- //resultLength := len(workflowExecution.Results)
- dbSave := false
- setExecution := true
-
- if actionResult.Action.ID == "" {
- //log.Printf("[ERROR] Failed handling EMPTY action %#v", actionResult)
+ //log.Printf("BASE LENGTH: %d", len(workflowExecution.Results))
+ workflowExecution, dbSave, err := shuffle.ParsedExecutionResult(ctx, *workflowExecution, actionResult)
+ if err != nil {
+ log.Printf("[ERROR] Failed execution of parsedexecution: %s", err)
resp.WriteHeader(401)
- resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Can't handle empty action"}`)))
+ resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`)))
return
}
- //tx, err := dbclient.NewTransaction(ctx)
- //if err != nil {
- // log.Printf("client.NewTransaction: %v", err)
- // resp.WriteHeader(401)
- // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed creating transaction"}`)))
- // return
- //}
- //key := datastore.NameKey("workflowexecution", workflowExecutionId, nil)
- //workflowExecution := &WorkflowExecution{}
- //if err := tx.Get(key, workflowExecution); err != nil {
- // log.Printf("[ERROR] tx.Get bug: %v", err)
- // tx.Rollback()
- // resp.WriteHeader(401)
- // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting the workflow key"}`)))
- // return
- //}
+ //log.Printf("NEW LENGTH: %d", len(workflowExecution.Results))
- if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" {
- dbSave = true
+ _ = dbSave
+ //resultLength := len(workflowExecution.Results)
+ setExecution := true
- newResults := []shuffle.ActionResult{}
- childNodes := []string{}
- if workflowExecution.Workflow.Configuration.ExitOnError {
- log.Printf("[WARNING] Actionresult is %s for node %s in %s. Should set workflowExecution and exit all running functions", actionResult.Status, actionResult.Action.ID, workflowExecution.ExecutionId)
- workflowExecution.Status = actionResult.Status
- workflowExecution.LastNode = actionResult.Action.ID
- // Find underlying nodes and add them
- } else {
- 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 {
- if nodeId == actionResult.Action.ID {
- continue
- }
-
- // 1. Find the action itself
- // 2. Create an actionresult
- curAction := shuffle.Action{ID: ""}
- for _, action := range workflowExecution.Workflow.Actions {
- if action.ID == nodeId {
- curAction = action
- break
- }
- }
-
- if len(curAction.ID) == 0 {
- log.Printf("Couldn't find subnode %s", nodeId)
- continue
- }
-
- resultExists := false
- for _, result := range workflowExecution.Results {
- if result.Action.ID == curAction.ID {
- resultExists = true
- break
- }
- }
-
- if !resultExists {
- // Check parents are done here. Only add it IF all parents are skipped
- skipNodeAdd := false
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.DestinationID == nodeId {
- // If the branch's source node is NOT in childNodes, it's not a skipped parent
- sourceNodeFound := false
- for _, item := range childNodes {
- if item == branch.SourceID {
- sourceNodeFound = true
- break
- }
- }
-
- if !sourceNodeFound {
- log.Printf("Not setting node %s to SKIPPED", nodeId)
- skipNodeAdd = true
- break
- }
- }
- }
-
- if !skipNodeAdd {
- newAction := shuffle.Action{
- AppName: curAction.AppName,
- AppVersion: curAction.AppVersion,
- Label: curAction.Label,
- Name: curAction.Name,
- ID: curAction.ID,
- }
-
- newResult := shuffle.ActionResult{
- Action: newAction,
- ExecutionId: actionResult.ExecutionId,
- Authorization: actionResult.Authorization,
- Result: "Skipped because of previous node",
- StartedAt: 0,
- CompletedAt: 0,
- Status: "SKIPPED",
- }
-
- newResults = append(newResults, newResult)
- //increaseStatisticsField(ctx, "workflow_execution_actions_skipped", workflowExecution.Workflow.ID, 1, workflowExecution.ExecutionOrg)
- }
- }
- }
- }
-
- // Cleans up aborted, and always gives a result
- lastResult := ""
- // type ActionResult struct {
- for _, result := range workflowExecution.Results {
- if actionResult.Action.ID == result.Action.ID {
- continue
- }
-
- if result.Status == "EXECUTING" {
- result.Status = actionResult.Status
- result.Result = "Aborted because of error in another node (2)"
- }
-
- if len(result.Result) > 0 {
- lastResult = result.Result
- }
-
- newResults = append(newResults, result)
- }
-
- workflowExecution.Result = lastResult
- workflowExecution.Results = newResults
-
- if workflowExecution.Status == "ABORTED" {
- //err = increaseStatisticsField(ctx, "workflow_executions_aborted", workflowExecution.Workflow.ID, 1, workflowExecution.ExecutionOrg)
- //if err != nil {
- // log.Printf("Failed to increase aborted execution stats: %s", err)
- //}
- } else if workflowExecution.Status == "FAILURE" {
- //err = increaseStatisticsField(ctx, "workflow_executions_failure", workflowExecution.Workflow.ID, 1, workflowExecution.ExecutionOrg)
- //if err != nil {
- // log.Printf("Failed to increase failure execution stats: %s", err)
- //}
- }
- }
-
- // FIXME rebuild to be like this or something
- // workflowExecution/ExecutionId/Nodes/NodeId
- // Find the appropriate action
- //log.Printf("[INFO] Setting value of %s in workflow %s to %s (1)", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status)
- if len(workflowExecution.Results) > 0 {
- // FIXME
- skip := false
- found := false
- outerindex := 0
- for index, item := range workflowExecution.Results {
- if item.Action.ID == actionResult.Action.ID {
- found = true
- if item.Status == actionResult.Status {
- skip = true
- }
-
- outerindex = index
- break
- }
- }
-
- if skip {
- //log.Printf("Both are %s. Skipping this node", item.Status)
- } else if found {
- // If result exists and execution variable exists, update execution value
- //log.Printf("Exec var backend: %s", workflowExecution.Results[outerindex].Action.ExecutionVariable.Name)
- actionVarName := workflowExecution.Results[outerindex].Action.ExecutionVariable.Name
- // Finds potential execution arguments
- if len(actionVarName) > 0 {
- log.Printf("EXECUTION VARIABLE LOCAL: %s", actionVarName)
- for index, execvar := range workflowExecution.ExecutionVariables {
- if execvar.Name == actionVarName {
- // Sets the value for the variable
- workflowExecution.ExecutionVariables[index].Value = actionResult.Result
- break
- }
- }
- }
-
- log.Printf("[INFO] Updating %s in workflow %s from %s to %s (3)", actionResult.Action.ID, workflowExecution.ExecutionId, workflowExecution.Results[outerindex].Status, actionResult.Status)
- workflowExecution.Results[outerindex] = actionResult
- } else {
- log.Printf("[INFO] Setting value of %s in workflow %s to %s (1)", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status)
- workflowExecution.Results = append(workflowExecution.Results, actionResult)
- }
- } else {
- log.Printf("[INFO] Setting value of %s in workflow %s to %s (2)", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status)
- workflowExecution.Results = append(workflowExecution.Results, actionResult)
- }
-
- // FIXME: Have a check for skippednodes and their parents
- for resultIndex, result := range workflowExecution.Results {
- if result.Status != "SKIPPED" {
- continue
- }
-
- // Checks if all parents are skipped or failed. Otherwise removes them from the results
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.DestinationID == result.Action.ID {
- for _, subresult := range workflowExecution.Results {
- if subresult.Action.ID == branch.SourceID {
- if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" {
- log.Printf("SUBRESULT PARENT STATUS: %s", subresult.Status)
- log.Printf("Should remove resultIndex: %d", resultIndex)
-
- workflowExecution.Results = append(workflowExecution.Results[:resultIndex], workflowExecution.Results[resultIndex+1:]...)
-
- break
- }
- }
- }
- }
- }
- }
-
- extraInputs := 0
- for _, trigger := range workflowExecution.Workflow.Triggers {
- if trigger.Name == "User Input" && trigger.AppName == "User Input" {
- extraInputs += 1
- } else if trigger.Name == "Shuffle Workflow" && trigger.AppName == "Shuffle Workflow" {
- extraInputs += 1
- }
- }
-
- //log.Printf("EXTRA: %d", extraInputs)
- //log.Printf("LENGTH: %d - %d", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extraInputs)
-
- if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions)+extraInputs {
- //log.Printf("\nIN HERE WITH RESULTS %d vs %d\n", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extraInputs)
- finished := true
- lastResult := ""
-
- // Doesn't have to be SUCCESS and FINISHED everywhere anymore.
- skippedNodes := false
- for _, result := range workflowExecution.Results {
- if result.Status == "EXECUTING" {
- finished = false
- break
- }
-
- // FIXME: Check if ALL parents are skipped or if its just one. Otherwise execute it
- if result.Status == "SKIPPED" {
- skippedNodes = true
-
- // Checks if all parents are skipped or failed. Otherwise removes them from the results
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.DestinationID == result.Action.ID {
- for _, subresult := range workflowExecution.Results {
- if subresult.Action.ID == branch.SourceID {
- if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" {
- //log.Printf("SUBRESULT PARENT STATUS: %s", subresult.Status)
- //log.Printf("Should remove resultIndex: %d", resultIndex)
- finished = false
- break
- }
- }
- }
- }
-
- if !finished {
- break
- }
- }
- }
-
- lastResult = result.Result
- }
-
- // FIXME: Handle skip nodes - change status?
- _ = skippedNodes
-
- if finished {
- dbSave = true
- log.Printf("[INFO] Execution of %s finished.", workflowExecution.ExecutionId)
- //log.Println("Might be finished based on length of results and everything being SUCCESS or FINISHED - VERIFY THIS. Setting status to finished.")
-
- workflowExecution.Result = lastResult
- workflowExecution.Status = "FINISHED"
- workflowExecution.CompletedAt = int64(time.Now().Unix())
- if workflowExecution.LastNode == "" {
- workflowExecution.LastNode = actionResult.Action.ID
- }
-
- //err = increaseStatisticsField(ctx, "workflow_executions_success", workflowExecution.Workflow.ID, 1, workflowExecution.ExecutionOrg)
- //if err != nil {
- // log.Printf("Failed to increase success execution stats: %s", err)
- //}
-
- // Handles extra statistics stuff when it's done
- // Does autocomplete magic with JSON
- handleExecutionStatistics(*workflowExecution)
- }
- }
-
- // FIXME - why isn't this how it works otherwise, wtf?
- //workflow, err := shuffle.GetWorkflow(workflowExecution.Workflow.ID)
- //newActions := []Action{}
- //for _, action := range workflowExecution.Workflow.Actions {
- // log.Printf("Name: %s, Env: %s", action.Name, action.Environment)
- //}
-
- tmpJson, err := json.Marshal(workflowExecution)
- if err == nil {
- if len(tmpJson) >= 1048487 {
- dbSave = true
- log.Printf("[ERROR] Result length is too long! Need to reduce result size")
-
- // Result string `json:"result" datastore:"result,noindex"`
- // Arbitrary reduction size
- maxSize := 500000
- newResults := []shuffle.ActionResult{}
- for _, item := range workflowExecution.Results {
- if len(item.Result) > maxSize {
- item.Result = "[ERROR] Result too large to handle (https://github.com/frikky/shuffle/issues/171)"
- }
-
- newResults = append(newResults, item)
- }
-
- workflowExecution.Results = newResults
- }
- }
-
- // Validating that action results hasn't changed
- // Handled using cachhing, so actually pretty fast
- cacheKey := fmt.Sprintf("workflowexecution-%s", workflowExecution.ExecutionId)
- cache, err := shuffle.GetCache(ctx, cacheKey)
- if err == nil {
- cacheData := []byte(cache.([]uint8))
- //log.Printf("CACHEDATA: %#v", cacheData)
- err = json.Unmarshal(cacheData, &workflowExecution)
- if err == nil {
- if attempts > 5 {
- //log.Printf("\n\nSkipping execution input - %d vs %d. Attempts: (%d)\n\n", len(parsedValue.Results), resultLength, attempts)
- }
-
- attempts += 1
- if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) {
- runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp)
- return
- }
- }
- }
-
- //if value, found := requestCache.Get(cacheKey); found {
- // parsedValue := value.(*shuffle.WorkflowExecution)
- // if len(parsedValue.Results) > 0 && len(parsedValue.Results) != resultLength {
+ //newExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId)
+ //if err == nil {
+ // //log.Printf("GOT GOOD EXECUTION CACHE FOR %s!", workflowExecution.ExecutionId)
+ // if len(newExecution.Results) > 0 && len(newExecution.Results) != resultLength {
// setExecution = false
// if attempts > 5 {
// //log.Printf("\n\nSkipping execution input - %d vs %d. Attempts: (%d)\n\n", len(parsedValue.Results), resultLength, attempts)
// }
// attempts += 1
- // if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) {
- // runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp)
- // return
- // }
+ // //if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) {
+ // // log.Printf("RUNNING AGAIN!!")
+ // // runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp)
+ // // return
// }
+ //} else {
+ // log.Printf("[WARNING] Failed getting cache for %s: %s", workflowExecution.ExecutionId, err)
//}
if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" {
- err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, dbSave)
+ err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, true)
+ //err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, dbSave)
if err != nil {
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err)))
return
}
+ //handleExecutionResult(ctx, *workflowExecution)
} else {
log.Printf("Skipping setexec with status %s", workflowExecution.Status)
}
- //ExecutionId
- // Transactions: https://cloud.google.com/datastore/docs/concepts/transactions#datastore-datastore-transactional-update-go
- // Prevents timing issues
- //if _, err := tx.Put(key, workflowExecution); err != nil {
- // log.Printf("[ERROR] tx.Put error: %v", err)
- // err = tx.Rollback()
- // if err != nil {
- // log.Printf("[ERROR] Rollback error (3): %s", err)
- // }
+ if resp != nil {
+ resp.WriteHeader(200)
+ resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
+ }
- // resp.WriteHeader(401)
- // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err)))
- // return
- //}
-
- //if _, err = tx.Commit(); err != nil {
- // err = tx.Rollback()
- // if err != nil {
- // log.Printf("[ERROR] Rollback error expected ? (1): %s", err)
- // }
-
- // if attempts >= 7 {
- // log.Printf("[ERROR] QUITTING: tx.Commit %d: %v", attempts, err)
-
- // workflowExecution.Status = "ABORTED"
- // shuffle.SetWorkflowExecution(ctx, *workflowExecution, true)
-
- // resp.WriteHeader(401)
- // resp.Write([]byte(`{"success": false}`))
- // return
- // }
-
- // if attempts > 3 {
- // log.Printf("[WARNING] tx.Commit %d: %v", attempts, err)
- // }
-
- // attempts += 1
- // runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp)
- // return
- //} else {
- // //if grpc.Code(err) == codes.Aborted {
- // // return nil, ErrConcurrentTransaction
- // //}
- // //t.id = nil // mark the transaction as expired
- //}
-
- resp.WriteHeader(200)
- resp.Write([]byte(fmt.Sprintf(`{"success": true}`)))
}
func JSONCheck(str string) bool {
@@ -2250,7 +1815,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request
}
//log.Printf("[INFO] New startnode: %s", workflowExecution.Start)
- childNodes := findChildNodes(workflowExecution, workflowExecution.Start)
+ childNodes := shuffle.FindChildNodes(workflowExecution, workflowExecution.Start)
topic := "workflows"
startFound := false
diff --git a/docker-compose.yml b/docker-compose.yml
index f9399d8c..b2de1a33 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -2,7 +2,7 @@ version: '3'
services:
frontend:
#build: ./frontend
- image: ghcr.io/frikky/shuffle-frontend:0.8.72
+ image: ghcr.io/frikky/shuffle-frontend:0.8.73
container_name: shuffle-frontend
hostname: shuffle-frontend
ports:
@@ -17,7 +17,7 @@ services:
- backend
backend:
#build: ./backend
- image: ghcr.io/frikky/shuffle-backend:0.8.72
+ image: ghcr.io/frikky/shuffle-backend:0.8.73
container_name: shuffle-backend
hostname: ${BACKEND_HOSTNAME}
# Here for debugging:
@@ -47,7 +47,7 @@ services:
- database
orborus:
#build: ./functions/onprem/orborus
- image: ghcr.io/frikky/shuffle-orborus:0.8.72
+ image: ghcr.io/frikky/shuffle-orborus:0.8.73
container_name: shuffle-orborus
hostname: shuffle-orborus
networks:
@@ -55,8 +55,8 @@ services:
volumes:
- /var/run/docker.sock:/var/run/docker.sock
environment:
- - SHUFFLE_APP_SDK_VERSION=0.8.60
- - SHUFFLE_WORKER_VERSION=0.8.72
+ - SHUFFLE_APP_SDK_VERSION=0.8.73
+ - SHUFFLE_WORKER_VERSION=0.8.73
- ORG_ID=${ORG_ID}
- ENVIRONMENT_NAME=${ENVIRONMENT_NAME}
- BASE_URL=http://${OUTER_HOSTNAME}:${BACKEND_PORT}
diff --git a/frontend/src/components/ParsedAction.jsx b/frontend/src/components/ParsedAction.jsx
index 6b8e4938..40590a4a 100644
--- a/frontend/src/components/ParsedAction.jsx
+++ b/frontend/src/components/ParsedAction.jsx
@@ -546,7 +546,7 @@ const ParsedAction = (props) => {
var disabled = false
var rows = "5"
var openApiHelperText = "This is an OpenAPI specific field"
- if (selectedApp.generated && selectedApp.activated && data.name === "body") {
+ if (selectedApp.generated && data.name === "body") {
const regex = /\${(\w+)}/g
const found = placeholder.match(regex)
if (found === null) {
diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx
index eb2cf843..9b018e62 100644
--- a/frontend/src/views/AngularWorkflow.jsx
+++ b/frontend/src/views/AngularWorkflow.jsx
@@ -5827,6 +5827,17 @@ const AngularWorkflow = (props) => {
+ {executionData.status === "EXECUTING" ?
+
+
+
+
+
+ : null}
{executionData.status !== undefined && executionData.status.length > 0 ?
diff --git a/frontend/src/views/Workflows.jsx b/frontend/src/views/Workflows.jsx
index b06b8a7d..bfca455f 100644
--- a/frontend/src/views/Workflows.jsx
+++ b/frontend/src/views/Workflows.jsx
@@ -561,7 +561,6 @@ const Workflows = (props) => {
const newId = uuid.v4()
if (trigger.trigger_type === "WEBHOOK") {
- //"http://localhost:5002/api/v1/hooks/webhook_db179eb0-cb0c-4d2a-9c4b-53d2625a5008"
const hookname = "webhook_"+newId
if (trigger.parameters.length === 2) {
trigger.parameters[0].value = referenceUrl+"webhook_"+trigger.id
@@ -574,12 +573,10 @@ const Workflows = (props) => {
for (var branchkey in data.branches) {
const branch = data.branches[branchkey]
if (branch.source_id === trigger.id) {
- //console.log("CHANGING SOURCE ID")
branch.source_id = newId
}
if (branch.destination_id === trigger.id) {
- //console.log("CHANGING DESTINATION ID")
branch.destination_id = newId
}
}
@@ -610,16 +607,18 @@ const Workflows = (props) => {
for (var branchkey in data.branches) {
const branch = data.branches[branchkey]
if (branch.source_id === data.actions[key].id) {
- //console.log("CHANGING SOURCE ID IN ACTION")
branch.source_id = newId
}
if (branch.destination_id === data.actions[key].id) {
- //console.log("CHANGING DESTINATION ID IN ACTION")
branch.destination_id = newId
}
}
+ if (data.actions[key].id === data.start) {
+ data.start = newId
+ }
+
//data.actions[key].environment = isCloud ? "cloud" : "Shuffle"
data.actions[key].environment = ""
data.actions[key].id = newId
@@ -629,7 +628,7 @@ const Workflows = (props) => {
if (data.workflow_variables !== null && data.workflow_variables !== undefined) {
for (var key in data.workflow_variables) {
const param = data.workflow_variables[key]
- if (param.name.includes("key") || param.name.includes("user") || param.name.includes("pass") || param.name.includes("api") || param.name.includes("auth") || param.name.includes("secret")|| param.name.includes("email")) {
+ if (param.name.includes("key") || param.name.includes("user") || param.name.includes("pass") || param.name.includes("api") || param.name.includes("auth") || param.name.includes("secret") || param.name.includes("email")) {
param.value = ""
param.is_valid = false
}
@@ -659,6 +658,7 @@ const Workflows = (props) => {
const exportWorkflow = (data) => {
let exportFileDefaultName = data.name+'.json';
data = sanitizeWorkflow(data)
+ return
// Add correct ID's for triggers
// Add mag
diff --git a/functions/onprem/orborus/build.sh b/functions/onprem/orborus/build.sh
index 08b121c2..47d0510e 100644
--- a/functions/onprem/orborus/build.sh
+++ b/functions/onprem/orborus/build.sh
@@ -1,5 +1,5 @@
NAME=shuffle-orborus
-VERSION=0.8.72
+VERSION=0.8.73
echo "Running docker build with $NAME:$VERSION"
#docker rmi frikky/shuffle:$NAME --force
diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod
index 2991cf7f..addb6aa1 100644
--- a/functions/onprem/orborus/go.mod
+++ b/functions/onprem/orborus/go.mod
@@ -3,13 +3,14 @@ module orborus
go 1.13
require (
+ github.com/Masterminds/semver v1.5.0 // indirect
github.com/Microsoft/go-winio v0.4.16 // indirect
github.com/containerd/containerd v1.4.3 // indirect
github.com/docker/distribution v2.7.1+incompatible // indirect
github.com/docker/docker v20.10.1+incompatible
github.com/docker/go-connections v0.4.0 // indirect
github.com/docker/go-units v0.4.0 // indirect
- github.com/frikky/shuffle-shared v0.0.23 // indirect
+ github.com/frikky/shuffle-shared v0.0.28
github.com/gogo/protobuf v1.3.1 // indirect
github.com/mackerelio/go-osstat v0.1.0
github.com/opencontainers/go-digest v1.0.0 // indirect
diff --git a/functions/onprem/orborus/go.sum b/functions/onprem/orborus/go.sum
index ada1bd1f..8a8e200b 100644
--- a/functions/onprem/orborus/go.sum
+++ b/functions/onprem/orborus/go.sum
@@ -41,6 +41,8 @@ cloud.google.com/go/storage v1.12.0/go.mod h1:fFLk2dp2oAhDz8QFKwqrjdJvxSp/W2g7ni
dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
+github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
+github.com/Masterminds/semver v1.5.0/go.mod h1:MB6lktGJrhw8PrUyiEoblNEGEQ+RzHPF078ddwwvV3Y=
github.com/Microsoft/go-winio v0.4.16 h1:FtSW/jqD+l4ba5iPBj9CODVtgfYAD8w2wS923g/cFDk=
github.com/Microsoft/go-winio v0.4.16/go.mod h1:XB6nPKklQyQ7GC9LdcBEcBl8PF76WugXOPRXwdLnMv0=
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -73,6 +75,8 @@ github.com/frikky/shuffle-shared v0.0.12 h1:+0EIfThmK47Po+LogPYZR4XjbS4Ds19WNMFu
github.com/frikky/shuffle-shared v0.0.12/go.mod h1:SEY432/xs4oBkOUnGwxiYKCZWZLRaheGInhAoW7N8ww=
github.com/frikky/shuffle-shared v0.0.23 h1:Pnlc2M6fHnFRLFd5K1iLTVv4/t4P04Ri1GJ5CMxzq0U=
github.com/frikky/shuffle-shared v0.0.23/go.mod h1:H7SqOta/EAYnfYuWzwzYSh/oWfF0kgnuaJTQNKQBvoQ=
+github.com/frikky/shuffle-shared v0.0.28 h1:VQqL3+ePwKSUxCOiCC8DpOEgbb2GhXI8XzFB/YlHbps=
+github.com/frikky/shuffle-shared v0.0.28/go.mod h1:H7SqOta/EAYnfYuWzwzYSh/oWfF0kgnuaJTQNKQBvoQ=
github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh
index 0c4d435e..4f9dc1df 100644
--- a/functions/onprem/worker/build.sh
+++ b/functions/onprem/worker/build.sh
@@ -1,5 +1,5 @@
NAME=shuffle-worker
-VERSION=0.8.72
+VERSION=0.8.73
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 0e54ab5a..73f41c23 100644
--- a/functions/onprem/worker/go.mod
+++ b/functions/onprem/worker/go.mod
@@ -3,16 +3,19 @@ module worker
go 1.15
require (
+ github.com/Masterminds/semver v1.5.0 // indirect
+ github.com/Microsoft/go-winio v0.4.16 // indirect
github.com/containerd/containerd v1.4.4 // indirect
github.com/docker/distribution v2.7.1+incompatible // indirect
- github.com/docker/docker v20.10.5+incompatible // indirect
+ 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/frikky/shuffle-shared v0.0.24 // indirect
+ github.com/frikky/shuffle-shared v0.0.28
github.com/gogo/protobuf v1.3.2 // indirect
- github.com/gorilla/mux v1.8.0 // indirect
+ github.com/gorilla/mux v1.8.0
github.com/opencontainers/go-digest v1.0.0 // indirect
github.com/opencontainers/image-spec v1.0.1 // indirect
+ github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/pkg/errors v0.9.1 // indirect
github.com/sirupsen/logrus v1.8.1 // indirect
)
diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go
index 91e5b975..5272b052 100644
--- a/functions/onprem/worker/worker.go
+++ b/functions/onprem/worker/worker.go
@@ -94,9 +94,9 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
sleepDuration := 1
if handleResultSend && requestsSent < 2 {
- data, err := json.Marshal(workflowExecution)
+ shutdownData, err := json.Marshal(workflowExecution)
if err == nil {
- sendResult(workflowExecution, data)
+ sendResult(workflowExecution, shutdownData)
log.Printf("[WARNING] Sent shutdown update")
} else {
log.Printf("[WARNING] DIDNT send update")
@@ -300,32 +300,25 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
ShowStdout: true,
}
+ exit := true
out, err := cli.ContainerLogs(ctx, cont.ID, logOptions)
if err != nil {
log.Printf("[INFO] Failed getting logs: %s", err)
} else {
- log.Printf("IN ELSE FOR DEPLOY")
buf := new(strings.Builder)
io.Copy(buf, out)
logs := buf.String()
- log.Printf("Logs: %s", logs)
+ log.Printf("Execution Logs: %s", logs)
- //log.Printf(logs)
- // check errors
- /*
- if strings.Contains(logs, "Error") {
- log.Printf("ERROR IN %s?", cont.ID)
- log.Println(logs)
- //return errors.New(fmt.Sprintf("ERROR FROM CONTAINER %s", cont.ID))
- } else {
- log.Printf("NORMAL EXEC OF %s?", cont.ID)
- }
- */
+ if strings.Contains(logs, "Normal execution.") {
+ exit = false
+ }
}
- log.Printf("ERROR IN CONTAINER DEPLOYMENT - ITS EXITED!")
-
- return errors.New(fmt.Sprintf(`{"success": false, "reason": "Container %s exited prematurely.","debug": "docker logs -f %s"}`, cont.ID, cont.ID))
+ if exit {
+ log.Printf("ERROR IN CONTAINER DEPLOYMENT - ITS EXITED!")
+ return errors.New(fmt.Sprintf(`{"success": false, "reason": "Container %s exited prematurely.","debug": "docker logs -f %s"}`, cont.ID, cont.ID))
+ }
}
}
}
@@ -519,6 +512,7 @@ func removeIndex(s []string, i int) []string {
}
func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
+ log.Printf("Inside execution results with %d / %d results", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions))
if len(startAction) == 0 {
startAction = workflowExecution.Start
if len(startAction) == 0 {
@@ -1243,7 +1237,7 @@ func executionInit(workflowExecution shuffle.WorkflowExecution) error {
return nil
}
-func handleExecution(client *http.Client, req *http.Request, workflowExecution shuffle.WorkflowExecution) error {
+func handleDefaultExecution(client *http.Client, req *http.Request, workflowExecution shuffle.WorkflowExecution) error {
// if no onprem runs (shouldn't happen, but extra check), exit
// if there are some, load the images ASAP for the app
@@ -1253,16 +1247,14 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution s
shutdown(workflowExecution, "", "", true)
}
- log.Printf("Startaction: %s", startAction)
+ log.Printf("DEFAULT EXECUTION Startaction: %s", startAction)
+
+ ctx := context.Background()
+ setWorkflowExecution(ctx, workflowExecution, false)
- // source = parent node, dest = child node
- // parent can have more children, child can have more parents
- // Process the parents etc. How?
for {
- handleExecutionResult(workflowExecution)
-
//fullUrl := fmt.Sprintf("%s/api/v1/workflows/%s/executions/%s/abort", baseUrl, workflowExecution.Workflow.ID, workflowExecution.ExecutionId)
- fullUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
+ fullUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl)
log.Printf("URL: %s", fullUrl)
req, err := http.NewRequest(
"POST",
@@ -1313,6 +1305,8 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution s
shutdown(workflowExecution, "", "", true)
}
+ setWorkflowExecution(ctx, workflowExecution, false)
+ //handleExecutionResult(workflowExecution)
}
return nil
@@ -1515,53 +1509,6 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
}
-func findChildNodes(workflowExecution shuffle.WorkflowExecution, nodeId string) []string {
- //log.Printf("\nNODE TO FIX: %s\n\n", nodeId)
- allChildren := []string{nodeId}
-
- // 1. Find children of this specific node
- // 2. Find the children of those nodes etc.
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.SourceID == nodeId {
- //log.Printf("Children: %s", branch.DestinationID)
- allChildren = append(allChildren, branch.DestinationID)
-
- childNodes := findChildNodes(workflowExecution, branch.DestinationID)
- for _, bottomChild := range childNodes {
- found := false
- for _, topChild := range allChildren {
- if topChild == bottomChild {
- found = true
- break
- }
- }
-
- if !found {
- allChildren = append(allChildren, bottomChild)
- }
- }
- }
- }
-
- // 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
-}
-
// Will make sure transactions are always ran for an execution. This is recursive if it fails. Allowed to fail up to 5 times
func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workflowExecutionId string, actionResult shuffle.ActionResult, resp http.ResponseWriter) {
//log.Printf("IN WORKFLOWEXECUTION SUB!")
@@ -1574,416 +1521,18 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl
return
}
- log.Printf(`[INFO] Got result %s from %s`, actionResult.Status, actionResult.Action.ID)
resultLength := len(workflowExecution.Results)
- dbSave := false
setExecution := true
- if len(actionResult.Action.ExecutionVariable.Name) > 0 {
- actionResult.Action.ExecutionVariable.Value = actionResult.Result
-
- foundIndex := -1
- for i, executionVariable := range workflowExecution.ExecutionVariables {
- if executionVariable.Name == actionResult.Action.ExecutionVariable.Name {
- foundIndex = i
- break
- }
- }
-
- if foundIndex >= 0 {
- workflowExecution.ExecutionVariables[foundIndex] = actionResult.Action.ExecutionVariable
- } else {
- workflowExecution.ExecutionVariables = append(workflowExecution.ExecutionVariables, actionResult.Action.ExecutionVariable)
- }
- }
-
- actionResult.Action = shuffle.Action{
- AppName: actionResult.Action.AppName,
- AppVersion: actionResult.Action.AppVersion,
- Label: actionResult.Action.Label,
- Name: actionResult.Action.Name,
- ID: actionResult.Action.ID,
- Parameters: actionResult.Action.Parameters,
- ExecutionVariable: actionResult.Action.ExecutionVariable,
- }
-
- if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" {
- //dbSave = true
-
- newResults := []shuffle.ActionResult{}
- childNodes := []string{}
- if workflowExecution.Workflow.Configuration.ExitOnError {
- log.Printf("[WARNING] shuffle.Actionresult is %s for node %s in %s. Should set workflowExecution and exit all running functions", actionResult.Status, actionResult.Action.ID, workflowExecution.ExecutionId)
- workflowExecution.Status = actionResult.Status
- workflowExecution.LastNode = actionResult.Action.ID
- // Find underlying nodes and add them
- } else {
- log.Printf("[WARNING] shuffle.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
- // Remove duplicates
- //log.Printf("CHILD NODES: %d", len(childNodes))
- childNodes = findChildNodes(*workflowExecution, actionResult.Action.ID)
- for _, nodeId := range childNodes {
- if nodeId == actionResult.Action.ID {
- continue
- }
-
- // 1. Find the action itself
- // 2. Create an actionresult
- curAction := shuffle.Action{ID: ""}
- for _, action := range workflowExecution.Workflow.Actions {
- if action.ID == nodeId {
- curAction = action
- break
- }
- }
-
- if len(curAction.ID) == 0 {
- log.Printf("Couldn't find subnode %s", nodeId)
- continue
- }
-
- resultExists := false
- for _, result := range workflowExecution.Results {
- if result.Action.ID == curAction.ID {
- resultExists = true
- break
- }
- }
-
- if !resultExists {
- // Check parents are done here. Only add it IF all parents are skipped
- skipNodeAdd := false
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.DestinationID == nodeId {
- // If the branch's source node is NOT in childNodes, it's not a skipped parent
- sourceNodeFound := false
- for _, item := range childNodes {
- if item == branch.SourceID {
- sourceNodeFound = true
- break
- }
- }
-
- if !sourceNodeFound {
- // FIXME: Shouldn't add skip for child nodes of these nodes. Check if this node is parent of upcoming nodes.
- log.Printf("\n\n NOT setting node %s to SKIPPED", nodeId)
- skipNodeAdd = true
-
- if !arrayContains(visited, nodeId) && !arrayContains(executed, nodeId) {
- nextActions = append(nextActions, nodeId)
- log.Printf("SHOULD EXECUTE NODE %s. Next actions: %s", nodeId, nextActions)
- }
- break
- }
- }
- }
-
- if !skipNodeAdd {
- newResult := shuffle.ActionResult{
- Action: curAction,
- ExecutionId: actionResult.ExecutionId,
- Authorization: actionResult.Authorization,
- Result: "Skipped because of previous node",
- StartedAt: 0,
- CompletedAt: 0,
- Status: "SKIPPED",
- }
-
- newResults = append(newResults, newResult)
- } else {
- //log.Printf("\n\nNOT adding %s as skipaction - should add to execute?", nodeId)
- //var visited []string
- //var executed []string
- //var nextActions []string
- }
- }
- }
- }
-
- // Cleans up aborted, and always gives a result
- lastResult := ""
- // type shuffle.ActionResult struct {
- for _, result := range workflowExecution.Results {
- if actionResult.Action.ID == result.Action.ID {
- continue
- }
-
- if result.Status == "EXECUTING" {
- result.Status = actionResult.Status
- result.Result = "Aborted because of error in another node (2)"
- }
-
- if len(result.Result) > 0 {
- lastResult = result.Result
- }
-
- newResults = append(newResults, result)
- }
-
- workflowExecution.Result = lastResult
- workflowExecution.Results = newResults
- }
-
- // FIXME rebuild to be like this or something
- // workflowExecution/ExecutionId/Nodes/NodeId
- // Find the appropriate action
- if len(workflowExecution.Results) > 0 {
- // FIXME
- skip := false
- found := false
- outerindex := 0
- for index, item := range workflowExecution.Results {
- if item.Action.ID == actionResult.Action.ID {
- found = true
-
- if item.Status == actionResult.Status {
- skip = true
- }
-
- outerindex = index
- break
- }
- }
-
- if skip {
- //log.Printf("Both are %s. Skipping this node", item.Status)
- } else if found {
- // If result exists and execution variable exists, update execution value
- //log.Printf("Exec var backend: %s", workflowExecution.Results[outerindex].Action.ExecutionVariable.Name)
- // Finds potential execution arguments
- actionVarName := workflowExecution.Results[outerindex].Action.ExecutionVariable.Name
- if len(actionVarName) > 0 {
- log.Printf("EXECUTION VARIABLE LOCAL: %s", actionVarName)
- for index, execvar := range workflowExecution.ExecutionVariables {
- if execvar.Name == actionVarName {
- // Sets the value for the variable
- workflowExecution.ExecutionVariables[index].Value = actionResult.Result
- break
- }
- }
- }
-
- log.Printf("[INFO] Updating %s in workflow %s from %s to %s", actionResult.Action.ID, workflowExecution.ExecutionId, workflowExecution.Results[outerindex].Status, actionResult.Status)
- workflowExecution.Results[outerindex] = actionResult
- } else {
- workflowExecution.Results = append(workflowExecution.Results, actionResult)
- log.Printf("[INFO] Setting value (1) of %s in execution %s to %s. New result length: %d", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status, len(workflowExecution.Results))
- }
- } else {
- workflowExecution.Results = append(workflowExecution.Results, actionResult)
- log.Printf("[INFO] Setting value (2) of %s in execution %s to %s. New result length: %d", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status, len(workflowExecution.Results))
- }
-
- if actionResult.Status == "SKIPPED" {
- log.Printf("\n\n[INFO] Handling special case for SKIPPED!\n\n")
- childNodes := findChildNodes(*workflowExecution, actionResult.Action.ID)
- for _, nodeId := range childNodes {
- if nodeId == actionResult.Action.ID {
- continue
- }
-
- // 1. Find the action itself
- // 2. Create an actionresult
- curAction := shuffle.Action{ID: ""}
- for _, action := range workflowExecution.Workflow.Actions {
- if action.ID == nodeId {
- curAction = action
- break
- }
- }
-
- if len(curAction.ID) == 0 {
- log.Printf("Couldn't find subnode %s", nodeId)
- continue
- }
-
- resultExists := false
- for _, result := range workflowExecution.Results {
- if result.Action.ID == curAction.ID {
- resultExists = true
- break
- }
- }
-
- if !resultExists {
- // Check parents are done here. Only add it IF all parents are skipped
- skipNodeAdd := false
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.DestinationID == nodeId {
- // If the branch's source node is NOT in childNodes, it's not a skipped parent
- sourceNodeFound := false
- for _, item := range childNodes {
- if item == branch.SourceID {
- sourceNodeFound = true
- break
- }
- }
-
- if !sourceNodeFound {
- log.Printf("[INFO] Not setting node %s to SKIPPED", nodeId)
- skipNodeAdd = true
- break
- }
- }
- }
-
- if !skipNodeAdd {
- newAction := shuffle.Action{
- AppName: curAction.AppName,
- AppVersion: curAction.AppVersion,
- Label: curAction.Label,
- Name: curAction.Name,
- ID: curAction.ID,
- }
-
- newResult := shuffle.ActionResult{
- Action: newAction,
- ExecutionId: actionResult.ExecutionId,
- Authorization: actionResult.Authorization,
- Result: "Skipped because of previous node",
- StartedAt: 0,
- CompletedAt: 0,
- Status: "SKIPPED",
- }
-
- workflowExecution.Results = append(workflowExecution.Results, newResult)
- }
- }
- }
- }
-
- // FIXME: Have a check for skippednodes and their parents
- /*
- for resultIndex, result := range workflowExecution.Results {
- if result.Status != "SKIPPED" {
- continue
- }
-
- // Checks if all parents are skipped or failed.
- // Otherwise removes them from the results
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.DestinationID == result.Action.ID {
- for _, subresult := range workflowExecution.Results {
- if subresult.Action.ID == branch.SourceID {
- if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" {
- //log.Printf("SUBRESULT PARENT STATUS: %s", subresult.Status)
- //log.Printf("Should remove resultIndex: %d", resultIndex)
-
- // FIXME: Reinstate this?
- //workflowExecution.Results = append(workflowExecution.Results[:resultIndex], workflowExecution.Results[resultIndex+1:]...)
- _ = resultIndex
-
- break
- }
- }
- }
- }
- }
- }
-
- log.Printf("NEW LENGTH: %d", len(workflowExecution.Results))
- */
-
- extraInputs := 0
- for _, trigger := range workflowExecution.Workflow.Triggers {
- if trigger.Name == "User Input" && trigger.AppName == "User Input" {
- extraInputs += 1
- } else if trigger.Name == "Shuffle Workflow" && trigger.AppName == "Shuffle Workflow" {
- extraInputs += 1
- }
- }
-
- //log.Printf("EXTRA: %d", extraInputs)
- //log.Printf("LENGTH: %d - %d", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extraInputs)
-
- if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions)+extraInputs {
- //log.Printf("\nIN HERE WITH RESULTS %d vs %d\n", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extraInputs)
- finished := true
- lastResult := ""
-
- // Doesn't have to be SUCCESS and FINISHED everywhere anymore.
- skippedNodes := false
- for _, result := range workflowExecution.Results {
- if result.Status == "EXECUTING" {
- finished = false
- break
- }
-
- // FIXME: Check if ALL parents are skipped or if its just one. Otherwise execute it
- if result.Status == "SKIPPED" {
- skippedNodes = true
-
- // Checks if all parents are skipped or failed. Otherwise removes them from the results
- for _, branch := range workflowExecution.Workflow.Branches {
- if branch.DestinationID == result.Action.ID {
- for _, subresult := range workflowExecution.Results {
- if subresult.Action.ID == branch.SourceID {
- if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" {
- //log.Printf("SUBRESULT PARENT STATUS: %s", subresult.Status)
- //log.Printf("Should remove resultIndex: %d", resultIndex)
- finished = false
- break
- }
- }
- }
- }
-
- if !finished {
- break
- }
- }
- }
-
- lastResult = result.Result
- }
-
- // FIXME: Handle skip nodes - change status?
- _ = skippedNodes
-
- if finished {
- dbSave = true
- log.Printf("[INFO] Execution of %s finished.", workflowExecution.ExecutionId)
- //log.Println("Might be finished based on length of results and everything being SUCCESS or FINISHED - VERIFY THIS. Setting status to finished.")
-
- workflowExecution.Result = lastResult
- workflowExecution.Status = "FINISHED"
- workflowExecution.CompletedAt = int64(time.Now().Unix())
- if workflowExecution.LastNode == "" {
- workflowExecution.LastNode = actionResult.Action.ID
- }
-
- }
- }
-
- // FIXME - why isn't this how it works otherwise, wtf?
- //workflow, err := getWorkflow(workflowExecution.Workflow.ID)
- //newActions := []Action{}
- //for _, action := range workflowExecution.Workflow.Actions {
- // log.Printf("Name: %s, Env: %s", action.Name, action.Environment)
- //}
-
- tmpJson, err := json.Marshal(workflowExecution)
- if err == nil {
- if len(tmpJson) >= 1048487 {
- dbSave = true
- log.Printf("[ERROR] Result length is too long! Need to reduce result size")
-
- // Result string `json:"result" datastore:"result,noindex"`
- // Arbitrary reduction size
- maxSize := 500000
- newResults := []shuffle.ActionResult{}
- for _, item := range workflowExecution.Results {
- if len(item.Result) > maxSize {
- item.Result = "[ERROR] Result too large to handle (https://github.com/frikky/shuffle/issues/171)"
- }
-
- newResults = append(newResults, item)
- }
-
- workflowExecution.Results = newResults
- }
+ workflowExecution, dbSave, err := shuffle.ParsedExecutionResult(ctx, *workflowExecution, actionResult)
+ if err != nil {
+ log.Printf("[ERROR] Failed execution of parsedexecution: %s", err)
+ resp.WriteHeader(401)
+ resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`)))
+ return
}
+ //log.Printf(`[INFO] Got result %s from %s`, actionResult.Status, actionResult.Action.ID)
+ //dbSave := false
if len(results) != len(workflowExecution.Results) {
log.Printf("\n\n[WARNING] There may have been an issue in transaction queue. Result lengths: %d vs %d. Should check which exists the base results, but not in entire execution, then append.\n\n", len(results), len(workflowExecution.Results))
@@ -2047,6 +1596,11 @@ func getWorkflowExecution(ctx context.Context, id string) (*shuffle.WorkflowExec
}
func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) {
+ if workflowExecution.ExecutionSource == "default" {
+ log.Printf("Not sending backend info since source is default")
+ return
+ }
+
fullUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl)
req, err := http.NewRequest(
"POST",
@@ -2083,13 +1637,13 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) {
//log.Printf("[FINISHED] Should send full result to %s", baseUrl)
//data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization)
- data, err := json.Marshal(workflowExecution)
+ shutdownData, err := json.Marshal(workflowExecution)
if err != nil {
log.Printf("[ERROR] Failed to unmarshal data for backend")
shutdown(workflowExecution, "", "", true)
}
- sendResult(workflowExecution, data)
+ sendResult(workflowExecution, shutdownData)
}
}
@@ -2153,6 +1707,7 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo
handleExecutionResult(workflowExecution)
validateFinished(workflowExecution)
+
if dbSave {
shutdown(workflowExecution, "", "", false)
}
@@ -2339,8 +1894,8 @@ func main() {
}
}
- log.Printf("Environments: %s. 1 = webserver, 0 or >1 = default", environments)
- if len(environments) == 1 { //&& workflowExecution.ExecutionSource != "default" {
+ 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)")
listener := webserverSetup(workflowExecution)
err := executionInit(workflowExecution)
@@ -2360,10 +1915,13 @@ func main() {
//wg.Add(1)
//wg.Wait()
} else {
- log.Printf("[INFO] Running NON-OPTIMIZED execution for type %s with %d environments", workflowExecution.ExecutionSource, len(environments))
-
+ 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)
+ //err := executionInit(workflowExecution)
+ //if err != nil {
+ // log.Printf("[INFO] Workflow setup failed: %s", workflowExecution.ExecutionId, err)
+ // shutdown(workflowExecution, "", "", true)
+ //}
}
-
}
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
@@ -2373,7 +1931,7 @@ func main() {
if workflowExecution.Status == "EXECUTING" || workflowExecution.Status == "RUNNING" {
//log.Printf("Status: %s", workflowExecution.Status)
- err = handleExecution(client, req, workflowExecution)
+ err = handleDefaultExecution(client, req, workflowExecution)
if err != nil {
log.Printf("[INFO] Workflow %s is finished: %s", workflowExecution.ExecutionId, err)
shutdown(workflowExecution, "", "", true)