From bb37e3e55d4d4062b3f3c98b1f3b344acad85230 Mon Sep 17 00:00:00 2001 From: Frikky Date: Thu, 7 Dec 2023 21:54:52 +0100 Subject: [PATCH] Minor play button fixes for worker + backend --- backend/go-app/go.mod | 2 +- backend/go-app/main.go | 3 + backend/go-app/walkoff.go | 183 +++++++++++++++++- frontend/src/components/ShuffleCodeEditor.jsx | 31 ++- frontend/src/views/AngularWorkflow.jsx | 81 +++----- 5 files changed, 235 insertions(+), 65 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index cbbc0ea3..af4b2f8b 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -18,7 +18,7 @@ require ( github.com/gorilla/mux v1.8.0 github.com/h2non/filetype v1.1.3 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.5.29 + github.com/shuffle/shuffle-shared v0.5.30 golang.org/x/crypto v0.14.0 google.golang.org/api v0.125.0 google.golang.org/grpc v1.55.0 diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 0d0f2f0f..571e8422 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -4801,6 +4801,7 @@ func initHandlers() { // App specific // From here down isnt checked for org specific r.HandleFunc("/api/v1/apps/{key}/execute", executeSingleAction).Methods("POST", "OPTIONS") + r.HandleFunc("/api/v1/apps/{key}/run", executeSingleAction).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/apps/categories", shuffle.GetActiveCategories).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/apps/categories/run", shuffle.RunCategoryAction).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/apps/upload", handleAppZipUpload).Methods("POST", "OPTIONS") @@ -4841,9 +4842,11 @@ func initHandlers() { r.HandleFunc("/api/v1/workflows/search", shuffle.HandleWorkflowRunSearch).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/workflows/schedules", shuffle.HandleGetSchedules).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/executions", shuffle.GetWorkflowExecutions).Methods("GET", "OPTIONS") + r.HandleFunc("/api/v1/workflows/{key}/executions/{key}/rerun", checkUnfinishedExecution).Methods("GET", "POST", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/executions/{key}/abort", shuffle.AbortExecution).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/schedule", scheduleWorkflow).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/workflows/download_remote", loadSpecificWorkflows).Methods("POST", "OPTIONS") + r.HandleFunc("/api/v1/workflows/{key}/run", executeWorkflow).Methods("GET", "POST", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/execute", executeWorkflow).Methods("GET", "POST", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/schedule/{schedule}", stopSchedule).Methods("DELETE", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/stream", shuffle.HandleStreamWorkflow).Methods("GET", "OPTIONS") diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 05ced567..dafbcdc7 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -1667,7 +1667,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request // FIXME - tmp name based on future companyname-companyId // This leads to issues with overlaps. Should set limits and such instead for _, environment := range execInfo.Environments { - log.Printf("[INFO] Execution: %s should execute onprem with execution environment \"%s\". Workflow: %s", workflowExecution.ExecutionId, environment, workflowExecution.Workflow.ID) + log.Printf("[INFO][%s] Execution: should execute onprem with execution environment \"%s\". Workflow: %s", workflowExecution.ExecutionId, environment, workflowExecution.Workflow.ID) executionRequest := shuffle.ExecutionRequest{ ExecutionId: workflowExecution.ExecutionId, @@ -3360,11 +3360,21 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) { return } - workflowExecution.Priority = 10 + + workflowExecution.Priority = 11 environments, err := shuffle.GetEnvironments(ctx, user.ActiveOrg.Id) environment := "Shuffle" if len(environments) >= 1 { + // Find default one environment = environments[0].Name + + for _, env := range environments { + if env.Default { + environment = env.Name + break + } + } + } else { log.Printf("[ERROR] No environments found for org %s. Exiting", user.ActiveOrg.Id) resp.WriteHeader(401) @@ -3372,6 +3382,14 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) { return } + // Enforcing same env for job + run to be default + // FIXME: Should use environment that is in the source workflow if it exists + for i, _ := range workflowExecution.Workflow.Actions { + workflowExecution.Workflow.Actions[i].Environment = environment + workflowExecution.Workflow.Actions[i].Label = "TMP" + } + shuffle.SetWorkflowExecution(ctx, workflowExecution, false) + log.Printf("[INFO] Execution (single action): %s should execute onprem with execution environment \"%s\". Workflow: %s", workflowExecution.ExecutionId, environment, workflowExecution.Workflow.ID) executionRequest := shuffle.ExecutionRequest{ @@ -3379,6 +3397,7 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) { WorkflowId: workflowExecution.Workflow.ID, Authorization: workflowExecution.Authorization, Environments: []string{environment}, + Priority: 11, } executionRequest.Priority = workflowExecution.Priority @@ -3399,6 +3418,9 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) { log.Printf("[ERROR] Failed to marshal retStruct in single execution: %s", err) } + // Deleting as this is a single action and doesn't need to be stored + shuffle.DeleteKey(ctx, "workflowexecution", executionRequest.ExecutionId) + resp.WriteHeader(200) resp.Write([]byte(returnBytes)) } @@ -3995,3 +4017,160 @@ func checkWorkflowApp(workflowApp shuffle.WorkflowApp) error { return nil } + +func checkUnfinishedExecution(resp http.ResponseWriter, request *http.Request) { + cors := shuffle.HandleCors(resp, request) + if cors { + return + } + + location := strings.Split(request.URL.String(), "/") + var fileId string + if location[1] == "api" { + if len(location) <= 4 { + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false}`)) + return + } + + fileId = location[4] + } + + if len(fileId) != 36 { + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false, "reason": "Workflow ID to abort is not valid"}`)) + return + } + + executionId := location[6] + if len(executionId) != 36 { + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false, "reason": "ExecutionID not valid"}`)) + return + } + + ctx := shuffle.GetContext(request) + exec, err := shuffle.GetWorkflowExecution(ctx, executionId) + if err != nil { + log.Printf("[ERROR] Failed getting execution (rerun workflow - 1) %s: %s", executionId, err) + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist (abort)."}`, executionId))) + return + } + + apikey := request.Header.Get("Authorization") + parsedKey := "" + if strings.HasPrefix(apikey, "Bearer ") { + apikeyCheck := strings.Split(apikey, " ") + if len(apikeyCheck) == 2 { + parsedKey = apikeyCheck[1] + } + } + + // ONLY allowed to run automatically with the same auth (july 2022) + if exec.Authorization != parsedKey { + user, err := shuffle.HandleApiAuthentication(resp, request) + if err != nil { + log.Printf("[ERROR][%s] Bad authorization key for execution (rerun workflow - 3): %s", executionId, err) + resp.WriteHeader(403) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed because you're not authorized to see this workflow (3)."}`))) + return + } + + // Check if user is in the correct org + if user.ActiveOrg.Id == exec.ExecutionOrg && user.Role != "org-reader" { + log.Printf("[AUDIT][%s] User %s (%s) is force continuing execution from org access", executionId, user.Username, user.Id) + } else if user.SupportAccess { + log.Printf("[AUDIT][%s] User %s (%s) is force continuing execution with support access", executionId, user.Username, user.Id) + } else { + log.Printf("[ERROR][%s] Bad authorization key for continue execution (rerun workflow - 2): %s", executionId, err) + resp.WriteHeader(403) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed because you're not authorized to see this workflow (2)."}`))) + return + } + } + + // Meant as a function that periodically checks whether previous executions have finished or not. + // Should probably be based on executedIds and finishedIds + // Schedule a check in the future instead? + + // Auth vs execution check! + extraInputs := 0 + for _, trigger := range exec.Workflow.Triggers { + if trigger.Name == "User Input" && trigger.AppName == "User Input" { + extraInputs += 1 + + //exec.Workflow.Actions = append(exec.Workflow.Actions, shuffle.Action{ + // ID: trigger.ID, + // Label: trigger.Label, + // Name: trigger.Name, + //}) + } else if trigger.Name == "Shuffle Workflow" && trigger.AppName == "Shuffle Workflow" { + extraInputs += 1 + + //exec.Workflow.Actions = append(exec.Workflow.Actions, shuffle.Action{ + // ID: trigger.ID, + // Label: trigger.Label, + // Name: trigger.Name, + //}) + } + } + + if exec.Status != "ABORTED" && exec.Status != "FINISHED" && exec.Status != "FAILURE" { + log.Printf("[DEBUG][%s] Rechecking execution and its status to send to backend IF the status is EXECUTING (%s - %d/%d finished)", exec.ExecutionId, exec.Status, len(exec.Results), len(exec.Workflow.Actions)+extraInputs) + } + + // Usually caused by issue during startup + if exec.Status == "" { + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No status for the execution"}`))) + return + } + + if exec.Status != "EXECUTING" { + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Already finished"}`))) + return + } + + // Force it back in the queue to be executed + if len(exec.Workflow.Actions) == 0 { + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Not a cloud env workflow. Only rerunning cloud env."}`))) + return + } + + log.Printf("[DEBUG][%s] Workflow: %s (%s)", exec.ExecutionId, exec.Workflow.Name, exec.Workflow.ID) + if exec.Workflow.ID == "" || exec.Workflow.Name == "" { + log.Printf("[ERROR][%s] No workflow ID found for execution", exec.ExecutionId) + shuffle.DeleteKey(ctx, "workflowexecution", exec.ExecutionId) + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "No workflow name / ID found. Can't run. Contact support@shuffler.io if this persists."}`))) + return + } + + environment := exec.Workflow.Actions[0].Environment + log.Printf("[DEBUG][%s] Not a cloud env workflow. Re-adding job in queue for env %s.", exec.ExecutionId, environment) + + parsedEnv := fmt.Sprintf("%s_%s", strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(environment, " ", "-"), "_", "-")), exec.ExecutionOrg) + log.Printf("[DEBUG][%s] Adding new run job to env (2): %s", exec.ExecutionId, parsedEnv) + + executionRequest := shuffle.ExecutionRequest{ + ExecutionId: exec.ExecutionId, + WorkflowId: exec.Workflow.ID, + Authorization: exec.Authorization, + Environments: []string{environment}, + } + + // Increase priority on reruns to catch up + executionRequest.Priority = 11 + err = shuffle.SetWorkflowQueue(ctx, executionRequest, parsedEnv) + if err != nil { + log.Printf("[ERROR] Failed adding execution to db: %s", err) + } + + + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Reran workflow in %s"}`, parsedEnv))) + +} diff --git a/frontend/src/components/ShuffleCodeEditor.jsx b/frontend/src/components/ShuffleCodeEditor.jsx index 3053083c..39b1a26c 100644 --- a/frontend/src/components/ShuffleCodeEditor.jsx +++ b/frontend/src/components/ShuffleCodeEditor.jsx @@ -1597,14 +1597,29 @@ const CodeEditor = (props) => { Expected Output - { - executeSingleAction(expOutput) - }}> - - {executing ? : } - - - + + + } diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index e9e28582..53a43232 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -11151,15 +11151,7 @@ const AngularWorkflow = (defaultprops) => { style={{ marginTop: 10 }} label={
Wait for results
} /> - -
+
{ // email,sms,app ... workflow.triggers[selectedTriggerIndex].parameters[2] = { name: "type", - value: "email", + value: "subflow", }; workflow.triggers[selectedTriggerIndex].parameters[3] = { @@ -12624,16 +12616,7 @@ const AngularWorkflow = (defaultprops) => { />
*/} - -
- Parameters +
{ display: "flex", }} > -
Information + + The information you want to show the user. Supports variables. +
{ display: "flex", }} > -
- Contact options + Input options + + Use subflows to connect to any app you want, or use the default email and sms options +
{ { + setTriggerOptionsWrapper("subflow"); + }} + color="primary" + value="subflow" + /> + } + label={
Subflow
} + /> + { 0 && workflow.triggers[selectedTriggerIndex].parameters[2] !== undefined && workflow.triggers[selectedTriggerIndex].parameters[2].value !== undefined ? workflow.triggers[selectedTriggerIndex].parameters[2].value.includes("sms") : false} onChange={() => { setTriggerOptionsWrapper("sms"); @@ -12728,21 +12714,8 @@ const AngularWorkflow = (defaultprops) => { } label={
SMS
} /> - { - setTriggerOptionsWrapper("subflow"); - }} - color="primary" - value="subflow" - /> - } - label={
Subflow
} - /> - {workflow.triggers[selectedTriggerIndex].parameters[2] !== undefined && workflow.triggers[selectedTriggerIndex].parameters[2].value.includes("subflow") ? ( + {workflow.triggers[selectedTriggerIndex].parameters[2] !== undefined && workflow.triggers[selectedTriggerIndex].parameters[2].value.includes("subflow") ? (
{workflows === undefined || workflows === null || @@ -12848,7 +12821,7 @@ const AngularWorkflow = (defaultprops) => { }, }} fullWidth - label="Email" + label="Email" color="primary" required placeholder={"mail1@company.com,mail2@company.com"}