From 2b639ceae38bd64645974a267d47ccb9b483fea5 Mon Sep 17 00:00:00 2001 From: Frikky Date: Wed, 27 Aug 2025 01:10:28 +0200 Subject: [PATCH] Increased timeouts and rerun mechanisms. Autofixes are now more and more in place by default --- backend/go-app/go.mod | 2 +- backend/go-app/main.go | 7 +++---- backend/go-app/walkoff.go | 21 +++++++++++++++------ functions/onprem/orborus/orborus.go | 13 ++++++++++++- functions/onprem/worker/worker.go | 1 + 5 files changed, 32 insertions(+), 12 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index 272235ed..ec880aaa 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -4,7 +4,7 @@ go 1.24.0 toolchain go1.24.3 -//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared +replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared //replace github.com/frikky/schemaless => ../../../schemaless //replace github.com/frikky/kin-openapi => ../../../../git/kin-openapi diff --git a/backend/go-app/main.go b/backend/go-app/main.go index a84462b4..c2088f84 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -5259,12 +5259,11 @@ func initHandlers() { r.HandleFunc("/api/v1/workflows/queue", handleGetWorkflowqueue).Methods("GET", "POST") r.HandleFunc("/api/v1/workflows/queue/confirm", handleGetWorkflowqueueConfirm).Methods("POST") - // 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") + // App specific. Partially Singul. r.HandleFunc("/api/v1/apps/categories", shuffle.GetActiveCategories).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/apps/categories/run", singul.RunCategoryAction).Methods("POST", "OPTIONS") + 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/run", shuffle.RunCategoryAction).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/apps/upload", handleAppZipUpload).Methods("POST", "OPTIONS") diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 38e767da..6840f79d 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -2857,24 +2857,24 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) { } location := strings.Split(request.URL.String(), "/") - var fileId string + var appId string if location[1] == "api" { if len(location) <= 4 { - resp.WriteHeader(401) + resp.WriteHeader(400) resp.Write([]byte(`{"success": false}`)) return } - fileId = location[4] + appId = location[4] } //log.Printf("[AUDIT] User Authentication failed in execute SINGLE action - CONTINUING ANYWAY: %s. Found OrgID: %#v", err, user.ActiveOrg.Id) - log.Printf("[AUDIT] User %s (%s) in org %s (%s) is running SINGLE App run for App ID '%s'", user.Username, user.Id, user.ActiveOrg.Name, user.ActiveOrg.Id, fileId) + log.Printf("[AUDIT] User %s (%s) in org %s (%s) is running SINGLE App run for App ID '%s'", user.Username, user.Id, user.ActiveOrg.Name, user.ActiveOrg.Id, appId) body, err := ioutil.ReadAll(request.Body) if err != nil { log.Printf("[INFO] Failed single execution POST body read: %s", err) - resp.WriteHeader(401) + resp.WriteHeader(400) resp.Write([]byte(`{"success": false}`)) return } @@ -2899,7 +2899,15 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) { decisionId = decision[0] } - workflowExecution, err := shuffle.PrepareSingleAction(ctx, user, fileId, body, runValidationAction, decisionId) + log.Printf("\n\nACTION TO RUN: %s. Body: %s. Source URL: %s\n\n", appId, string(body), request.URL.String()) + + workflowExecution, err := shuffle.PrepareSingleAction(ctx, user, appId, body, runValidationAction, decisionId) + if appId == "agent_starter" { + log.Printf("[INFO] Returning early for agent_starter single action execution: %s", workflowExecution.ExecutionId) + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "execution_id": "%s", "authorization": "%s"}`, workflowExecution.ExecutionId, workflowExecution.Authorization))) + return + } debugUrl := fmt.Sprintf("/workflows/%s?execution_id=%s", workflowExecution.Workflow.ID, workflowExecution.ExecutionId) resp.Header().Add("X-Debug-Url", debugUrl) @@ -2940,6 +2948,7 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) { go shuffle.IncrementCache(ctx, workflowExecution.OrgId, "workflow_executions") executionRequest := shuffle.ExecutionRequest{ + Priority: 15, ExecutionId: workflowExecution.ExecutionId, WorkflowId: workflowExecution.Workflow.ID, Authorization: workflowExecution.Authorization, diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 8bc23b0a..320382a5 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -2057,6 +2057,11 @@ func main() { log.Printf("[DEBUG] Verbose mode. NOT cleaning up. Cleanup env: %s", cleanupEnv) } + // Default to 120 instead of default 30 + if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) == 0 { + os.Setenv("SHUFFLE_APP_SDK_TIMEOUT", "120") + } + workerTimeout := 600 if workerTimeoutEnv != "" { tmpInt, err := strconv.Atoi(workerTimeoutEnv) @@ -3951,7 +3956,13 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string, } } - client := &http.Client{} + client := &http.Client{ + //Transport: &http.Transport{ + // TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + //}, + Timeout: time.Duration(120 * time.Second), + } + req, err := http.NewRequest( "POST", streamUrl, diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index b06a2fc0..67be9c38 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -4298,6 +4298,7 @@ func main() { if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" { logsDisabled = "true" } + /*** ENDREMOVE ***/ // Elasticsearch necessary to ensure we'ren ot running with Datastore configurations for minimal/maximal data sizes // Recursive import kind of :)