From 958d42998c6c5e1f072d9056a6ab450c940c7376 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Tue, 15 Oct 2024 22:46:01 +0530 Subject: [PATCH] fix: worker sending in wrong executionId --- functions/onprem/worker/worker.go | 46 ++++++++++++++----------------- 1 file changed, 21 insertions(+), 25 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 3402ea6f..190694fd 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -38,9 +38,9 @@ import ( "github.com/gorilla/mux" //k8s deps + appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - appsv1 "k8s.io/api/apps/v1" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/kubernetes" ) @@ -140,16 +140,10 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo err = shuffle.SetCache(ctx, cacheKey, execData, 30) if err != nil { - log.Printf("[ERROR][%s] Failed adding to cache during setexecution", workflowExecution) + log.Printf("[ERROR][%s] Failed adding to cache during setexecution", workflowExecution.ExecutionId) return err } - /*** STARTREMOVE ***/ - if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" { - return nil - } - /*** ENDREMOVE ***/ - handleExecutionResult(workflowExecution) validated := shuffle.ValidateFinished(ctx, -1, workflowExecution) if validated { @@ -532,7 +526,7 @@ func deployk8sApp(image string, identifier string, env []string) error { // } // use deployment instead of pod - // then expose a service similarly. + // then expose a service similarly. // number of replicas can be set to os.Getenv("SHUFFLE_SCALE_REPLICAS") replicaNumberStr := os.Getenv("SHUFFLE_SCALE_REPLICAS") replicaNumber := 1 @@ -542,7 +536,7 @@ func deployk8sApp(image string, identifier string, env []string) error { log.Printf("[ERROR] %s is not a valid number for replication", replicaNumberStr) } else { replicaNumber = tmpInt - + } } @@ -2902,8 +2896,6 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener { return listener } - - func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) { ctx := context.Background() nodes, err := dockercli.NodeList(ctx, types.NodeListOptions{}) @@ -3792,6 +3784,8 @@ func checkUnfinished(resp http.ResponseWriter, request *http.Request, execReques exec, err := shuffle.GetWorkflowExecution(ctx, execRequest.ExecutionId) log.Printf("[DEBUG][%s] Rechecking execution and it's status to send to backend IF the status is EXECUTING (%s - %d/%d finished)", execRequest.ExecutionId, exec.Status, len(exec.Results), len(exec.Workflow.Actions)) if err != nil { + log.Printf("[ERROR][%s] Got error: %s", execRequest.ExecutionId, err) + handleRunExecution(resp, request) return } @@ -3834,12 +3828,7 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) { resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) return } - - // Checks if a workflow is done 30 seconds later, and sends info to backend no matter what - go func() { - time.Sleep(time.Duration(30) * time.Second) - checkUnfinished(resp, request, execRequest) - }() + ctx := context.Background() // FIXME: This should be PER EXECUTION //if strings.ToLower(os.Getenv("SHUFFLE_PASS_APP_PROXY")) == "true" { @@ -3880,13 +3869,16 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) { } var workflowExecution shuffle.WorkflowExecution - data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, execRequest.ExecutionId, execRequest.Authorization) streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl) req, err := http.NewRequest( "POST", streamResultUrl, - bytes.NewBuffer([]byte(data)), + bytes.NewBuffer([]byte(fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, execRequest.ExecutionId, execRequest.Authorization))), ) + if err != nil { + log.Printf("[ERROR][%s] Failed to create a new request", execRequest.ExecutionId) + return + } client := shuffle.GetExternalClient(streamResultUrl) newresp, err := client.Do(req) @@ -3900,14 +3892,14 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) { defer newresp.Body.Close() body, err = ioutil.ReadAll(newresp.Body) if err != nil { - log.Printf("[ERROR] Failed reading body (2): %s", err) + log.Printf("[ERROR][%s] Failed reading body (2): %s", execRequest.ExecutionId, err) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) return } if newresp.StatusCode != 200 { - log.Printf("[ERROR] Bad statuscode: %d, %s", newresp.StatusCode, string(body)) + log.Printf("[ERROR][%s] Bad statuscode: %d, %s", execRequest.ExecutionId, newresp.StatusCode, string(body)) if strings.Contains(string(body), "Workflowexecution is already finished") { log.Printf("[DEBUG] Shutting down (19)") @@ -3927,13 +3919,18 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) { return } - ctx := context.Background() //err = shuffle.SetWorkflowExecution(ctx, workflowExecution, true) err = setWorkflowExecution(ctx, workflowExecution, true) if err != nil { log.Printf("[ERROR] Failed initializing execution saving for %s: %s", workflowExecution.ExecutionId, err) } + // Checks if a workflow is done 30 seconds later, and sends info to backend no matter what + go func() { + time.Sleep(time.Duration(30) * time.Second) + checkUnfinished(resp, request, execRequest) + }() + if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" { log.Printf("[DEBUG] Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId) log.Printf("[DEBUG] Shutting down (20)") @@ -3978,7 +3975,7 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) { err = executionInit(workflowExecution) if err != nil { - log.Printf("[DEBUG][%s] Shutting down (30) - Workflow setup failed: %s", workflowExecution.ExecutionId, workflowExecution.ExecutionId, err) + log.Printf("[DEBUG][%s] Shutting down (30) - Workflow setup failed: %s", workflowExecution.ExecutionId, err) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Error in execution init: %s"}`, err))) return @@ -4091,7 +4088,6 @@ func runWebserver(listener net.Listener) { //log.Fatal(http.Serve(listener, nil)) - log.Printf("[DEBUG] NEW webserver setup") http.Handle("/", r)