From 63ce93500c8d11868a342f95fe1424e3913172ba Mon Sep 17 00:00:00 2001 From: frikky Date: Fri, 8 Apr 2022 01:50:09 +0200 Subject: [PATCH] Fixed a broken part of the condition system which didn't look for skipped/failed nodes --- backend/app_sdk/app_base.py | 21 +++++++++++ backend/go-app/main.go | 2 +- functions/onprem/orborus/orborus.go | 55 ++++++++++++++++++++--------- 3 files changed, 61 insertions(+), 17 deletions(-) diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index 197db19c..82a14724 100644 --- a/backend/app_sdk/app_base.py +++ b/backend/app_sdk/app_base.py @@ -2466,6 +2466,23 @@ class AppBase: continue matching_branches += 1 + + # Find if previous is skipped or failed. Skipped != correct branch + try: + should_skip = False + for res in fullexecution["results"]: + if res["action"]["id"] == branch["source_id"]: + if res["status"] == "FAILURE" or res["status"] == "SKIPPED": + should_skip = True + + break + + if should_skip: + continue + except Exception as e: + self.logger.info("[WARNING] Failed handling check of if parent is skipped") + + # Remove anything without a condition try: if (branch["conditions"]) == 0 or branch["conditions"] == None: @@ -2475,6 +2492,8 @@ class AppBase: correct_branches += 1 continue + # FIXME: Check if the previous node has a result or not + self.logger.info("[DEBUG] Relevant conditions: %s" % branch["conditions"]) successful_conditions = [] failed_conditions = [] @@ -2535,6 +2554,8 @@ class AppBase: if matching_branches > 0 and correct_branches > 0: return True, "" + # FIXME: Check if previous branches are at all finished + self.logger.info("[DEBUG] Correct branches vs matching branches: %d vs %d" % (correct_branches, matching_branches)) return False, {"success": False, "reason": "Minimum of one branch's conditions must be correct to continue. Total: %d of %d" % (correct_branches, matching_branches)} diff --git a/backend/go-app/main.go b/backend/go-app/main.go index ec7faacb..b49817da 100644 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -4075,7 +4075,7 @@ func runInitEs(ctx context.Context) { // FIXME: Have this for all envs in all orgs (loop and find). if len(parsedApikey) > 0 { - cleanupSchedule := 600 + cleanupSchedule := 300 environments := []string{"Shuffle"} log.Printf("[DEBUG] Starting schedule setup for execution cleanup every %d seconds. Running first immediately.", cleanupSchedule) cleanupJob := func() func() { diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 5e21f808..d72df574 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -495,23 +495,24 @@ func deployWorker(image string, identifier string, env []string, executionReques // FIXME: Should we handle replies properly? // In certain cases, a workflow may e.g. be aborted already. If it's aborted, that returns // a 401 from the worker, which returns an error here - err := sendWorkerRequest(executionRequest) - if err != nil { - log.Printf("[ERROR] Failed worker request for %s: %s", executionRequest.ExecutionId, err) + go sendWorkerRequest(executionRequest) + //err := sendWorkerRequest(executionRequest) + //if err != nil { + // log.Printf("[ERROR] Failed worker request for %s: %s", executionRequest.ExecutionId, err) - if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") { - workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) - deployServiceWorkers(workerImage) + // if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") { + // workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) + // deployServiceWorkers(workerImage) - time.Sleep(time.Duration(10) * time.Second) - err = sendWorkerRequest(executionRequest) - } - } + // time.Sleep(time.Duration(10) * time.Second) + // err = sendWorkerRequest(executionRequest) + // } + //} - if err == nil { - // FIXME: Readd this? Removed for rerun reasons - // executionIds = append(executionIds, executionRequest.ExecutionId) - } + //if err == nil { + // // FIXME: Readd this? Removed for rerun reasons + // // executionIds = append(executionIds, executionRequest.ExecutionId) + //} //}() return nil @@ -1327,21 +1328,36 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", parsedBaseurl) } + client := &http.Client{} req, err := http.NewRequest( "POST", streamUrl, bytes.NewBuffer([]byte(data)), ) - - client := &http.Client{} if err != nil { log.Printf("[ERROR] Failed creating worker request: %s", err) + if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") { + workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) + deployServiceWorkers(workerImage) + + time.Sleep(time.Duration(10) * time.Second) + //err = sendWorkerRequest(executionRequest) + } + return err } newresp, err := client.Do(req) if err != nil { log.Printf("[ERROR] Error running worker request: %s", err) + if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") { + workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) + deployServiceWorkers(workerImage) + + time.Sleep(time.Duration(10) * time.Second) + //err = sendWorkerRequest(executionRequest) + } + return err } @@ -1353,6 +1369,13 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { if newresp.StatusCode != 200 { log.Printf("[ERROR] Error running worker request - status code is %d, not 200. Body: %s", newresp.StatusCode, string(body)) + + workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) + deployServiceWorkers(workerImage) + + time.Sleep(time.Duration(10) * time.Second) + //err = sendWorkerRequest(executionRequest) + return errors.New(fmt.Sprintf("Bad statuscode from worker: %d - expecting 200", newresp.StatusCode)) }