diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 8b90537d..156e476c 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -1464,7 +1464,7 @@ const AngularWorkflow = (props) => { // break // } } else { - console.log("INSIDE LAST CHECK: ", edge) + //console.log("INSIDE LAST CHECK: ", edge) // Find the targetnode and check if its a trigger // FIXME - do this for both actions and other types? @@ -2871,9 +2871,9 @@ const AngularWorkflow = (props) => { ) } - const runSearch = (event) => { - if (event.target.value.length > 0) { - setVisibleApps(allApps.filter(app => app.name.toLowerCase().includes(event.target.value.trim().toLowerCase()))) + const runSearch = (value) => { + if (value.length > 0) { + setVisibleApps(allApps.filter(app => app.name.toLowerCase().includes(value.trim().toLowerCase()))) } else { setVisibleApps(prioritizedApps.concat(filteredApps.filter(innerapp => !internalIds.includes(innerapp.id)))) } @@ -2904,8 +2904,14 @@ const AngularWorkflow = (props) => { color="primary" placeholder={"Search Active Apps"} id="appsearch" + onKeyPress={(event) => { + if (event.key === "Enter") { + console.log("ENTER!") + runSearch(event.target.value) + } + }} onBlur={(event) => { - runSearch(event) + runSearch(event.target.value) }} /> {visibleApps.length > 0 ? @@ -7253,7 +7259,7 @@ const AngularWorkflow = (props) => {
{data.action.label}
- + {data.action.name}
@@ -7261,7 +7267,7 @@ const AngularWorkflow = (props) => {
- Status   + Status  {data.status} diff --git a/functions/onprem/orborus/build.sh b/functions/onprem/orborus/build.sh index c6df2439..08b121c2 100644 --- a/functions/onprem/orborus/build.sh +++ b/functions/onprem/orborus/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-orborus -VERSION=0.8.71 +VERSION=0.8.72 echo "Running docker build with $NAME:$VERSION" #docker rmi frikky/shuffle:$NAME --force diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index a3fd36dd..c2cafa0f 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -248,7 +248,7 @@ func initializeImages() { log.Printf("[WARNING] SHUFFLE_APP_SDK_VERSION not defined. Defaulting to %s", appSdkVersion) } if workerVersion == "" { - workerVersion = "0.8.70" + workerVersion = "0.8.72" log.Printf("[WARNING] SHUFFLE_WORKER_VERSION not defined. Defaulting to %s", workerVersion) } diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 10f2c70c..0c4d435e 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-worker -VERSION=0.8.71 +VERSION=0.8.72 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 44d51077..0e54ab5a 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -8,7 +8,7 @@ require ( github.com/docker/docker v20.10.5+incompatible // indirect 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.20 // indirect + github.com/frikky/shuffle-shared v0.0.24 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/gorilla/mux v1.8.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 642af1bc..91e5b975 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -53,6 +53,7 @@ var nextActions []string var containerIds []string var extra int var startAction string +var results []shuffle.ActionResult var containerId string @@ -264,7 +265,10 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] ) if err != nil { - log.Printf("[WARNING] Container CREATE error: %s", err) + if !strings.Contains(err.Error(), "Conflict. The container name") { + log.Printf("[ERROR] Container CREATE error: %s", err) + } + return err } @@ -963,7 +967,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { pullOptions := types.ImagePullOptions{} if cleanupEnv == "true" { err = deployApp(dockercli, images[0], identifier, env, workflowExecution) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if strings.Contains(err.Error(), "exited prematurely") { shutdown(workflowExecution, action.ID, err.Error(), true) } @@ -978,7 +982,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { buildBuf := new(strings.Builder) _, err = io.Copy(buildBuf, reader) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { log.Printf("[ERROR] Error in IO copy: %s", err) shutdown(workflowExecution, action.ID, err.Error(), true) } else { @@ -991,7 +995,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } err = deployApp(dockercli, image, identifier, env, workflowExecution) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist") if strings.Contains(err.Error(), "exited prematurely") { @@ -1008,7 +1012,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } else { err = deployApp(dockercli, images[0], identifier, env, workflowExecution) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if strings.Contains(err.Error(), "exited prematurely") { shutdown(workflowExecution, action.ID, err.Error(), true) } @@ -1021,7 +1025,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } err = deployApp(dockercli, image, identifier, env, workflowExecution) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if strings.Contains(err.Error(), "exited prematurely") { shutdown(workflowExecution, action.ID, err.Error(), true) } @@ -1032,14 +1036,14 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } err = deployApp(dockercli, image, identifier, env, workflowExecution) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if strings.Contains(err.Error(), "exited prematurely") { shutdown(workflowExecution, action.ID, err.Error(), true) } log.Printf("[WARNING] Failed deploying image THREE TIMES. Attempting to download the latter as last resort.") reader, err := dockercli.ImagePull(context.Background(), image, pullOptions) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { log.Printf("[ERROR] Failed getting %s. The couldn't be find locally, AND is missing.", image) shutdown(workflowExecution, action.ID, err.Error(), true) } @@ -1059,7 +1063,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } err = deployApp(dockercli, image, identifier, env, workflowExecution) - if err != nil { + if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist") if strings.Contains(err.Error(), "exited prematurely") { shutdown(workflowExecution, action.ID, err.Error(), true) @@ -1120,6 +1124,8 @@ func executionInit(workflowExecution shuffle.WorkflowExecution) error { parents = map[string][]string{} children = map[string][]string{} + results = workflowExecution.Results + startAction = workflowExecution.Start log.Printf("[INFO] STARTACTION: %s", startAction) if len(startAction) == 0 { @@ -1501,57 +1507,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { } } - //if actionResult.Status == "WAITING" && actionResult.Action.AppName == "User Input" { - // log.Printf("SHOULD WAIT A BIT AND RUN CLOUD STUFF WITH USER INPUT! WAITING!") - - // var trigger shuffle.Trigger - // err = json.Unmarshal([]byte(actionResult.Result), &trigger) - // if err != nil { - // log.Printf("Failed unmarshaling actionresult for user input: %s", err) - // resp.WriteHeader(401) - // resp.Write([]byte(`{"success": false}`)) - // return - // } - - // orgId := workflowExecution.ExecutionOrg - // if len(workflowExecution.OrgId) == 0 && len(workflowExecution.Workflow.OrgId) > 0 { - // orgId = workflowExecution.Workflow.OrgId - // } - - // err := handleUserInput(trigger, orgId, workflowExecution.Workflow.ID, workflowExecution.ExecutionId) - // if err != nil { - // log.Printf("Failed userinput handler: %s", err) - // actionResult.Result = fmt.Sprintf("Cloud error: %s", err) - // workflowExecution.Results = append(workflowExecution.Results, actionResult) - // workflowExecution.Status = "ABORTED" - // err = setshuffle.WorkflowExecution(ctx, *workflowExecution, true) - // if err != nil { - // log.Printf("Failed ") - // } else { - // log.Printf("Successfully set the execution to waiting.") - // } - - // resp.WriteHeader(401) - // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Error: %s"}`, err))) - // } else { - // log.Printf("Successful userinput handler") - // resp.WriteHeader(200) - // resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "CLOUD IS DONE"}`))) - - // actionResult.Result = "Waiting for user feedback based on configuration" - - // workflowExecution.Results = append(workflowExecution.Results, actionResult) - // workflowExecution.Status = actionResult.Status - // err = setshuffle.WorkflowExecution(ctx, *workflowExecution, true) - // if err != nil { - // log.Printf("Failed ") - // } else { - // log.Printf("Successfully set the execution to waiting.") - // } - // } - - // return - //} + results = append(results, actionResult) resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": true}`))) @@ -1622,30 +1578,33 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl resultLength := len(workflowExecution.Results) dbSave := false setExecution := true - //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 := &shuffle.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 - //} + 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, + 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" { @@ -2026,6 +1985,10 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl } } + 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)) + } + // Validating that action results hasn't changed // Handled using cachhing, so actually pretty fast cacheKey := fmt.Sprintf("workflowexecution-%s", workflowExecution.ExecutionId)