From b66d81178a00fbce68ee67aca05ef74e7c505f63 Mon Sep 17 00:00:00 2001 From: frikky Date: Thu, 7 Apr 2022 21:44:42 +0200 Subject: [PATCH] Fixed further execution concurrency issues with sdk & worker --- backend/app_sdk/app_base.py | 61 ++++++++++++++++---------- backend/app_sdk/build.sh | 10 ++--- backend/go-app/docker.go | 10 ++--- backend/go-app/go.mod | 2 +- backend/go-app/walkoff.go | 22 +++++++--- docker-compose.yml | 11 +++-- frontend/src/views/AngularWorkflow.jsx | 24 +++++++++- functions/onprem/orborus/build.sh | 2 +- functions/onprem/orborus/go.mod | 2 +- functions/onprem/orborus/go.sum | 2 + functions/onprem/orborus/orborus.go | 4 +- functions/onprem/worker/build.sh | 2 +- functions/onprem/worker/go.mod | 2 +- functions/onprem/worker/worker.go | 13 +++--- 14 files changed, 108 insertions(+), 59 deletions(-) diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index 52cc5e83..197db19c 100644 --- a/backend/app_sdk/app_base.py +++ b/backend/app_sdk/app_base.py @@ -341,20 +341,28 @@ class AppBase: else: self.logger.info(f"[DEBUG] RESP: {ret.text}") - except (requests.exceptions.RequestException, TimeoutError) as e: - time.sleep(5) + except requests.exceptions.RequestException as e: + self.logger.info(f"[DEBUG] Request problem: {e}") + #time.sleep(5) + continue + except TimeoutError as e: + self.logger.info(f"[DEBUG] Timeout or request: {e}") + #time.sleep(5) continue except requests.exceptions.ConnectionError as e: - time.sleep(5) + self.logger.info(f"[DEBUG] Connectionerror: {e}") + #time.sleep(5) continue except http.client.RemoteDisconnected as e: - time.sleep(5) + self.logger.info(f"[DEBUG] Remote: {e}") + #time.sleep(5) continue except urllib3.exceptions.ProtocolError as e: - time.sleep(5) + self.logger.info(f"[DEBUG] Protocol err: {e}") + #time.sleep(5) continue - time.sleep(5) + #time.sleep(5) if not finished: # Not sure why this would work tho :) @@ -2516,7 +2524,7 @@ class AppBase: # self.logger.info("Failed condition check for %s %s %s." % (sourcevalue, condition["condition"]["value"], destinationvalue)) # return False, {"success": False, "reason": "Failed condition (3): %s %s %s" % (sourcevalue, condition["condition"]["value"], destinationvalue)} - self.logger.info("CONDITIONS VS SUCCESS: %d vs %d" % (total_conditions, successful_conditions)) + #self.logger.info("CONDITIONS VS SUCCESS: %d vs %d" % (total_conditions, successful_conditions)) if total_conditions == successful_conditions: correct_branches += 1 @@ -3347,7 +3355,7 @@ class AppBase: port = int(exposed_port) logger.info(f"[DEBUG] Starting webserver on port {port} (same as exposed port)") from flask import Flask, request - #from waitress import serve + from waitress import serve flask_app = Flask(__name__) #flask_app.config['PERMANENT_SESSION_LIFETIME'] = datetime.timedelta(minutes=5) @@ -3377,6 +3385,7 @@ class AppBase: #print(f"APP: {app}") app = cls(redis=None, logger=logger, console_logger=logger) + extra_info = "" try: #asyncio.run(AppBase.run(action=requestdata), debug=True) #value = json.dumps(value) @@ -3384,16 +3393,20 @@ class AppBase: app.full_execution = json.dumps(requestdata["workflow_execution"]) except Exception as e: logger.info(f"[ERROR] Failed parsing full execution from workflow_execution: {e}") + extra_info += f"\n{e}" + try: app.action = requestdata["action"] except Exception as e: logger.info(f"[ERROR] Failed parsing action: {e}") + extra_info += f"\n{e}" try: app.authorization = requestdata["authorization"] app.current_execution_id = requestdata["execution_id"] except Exception as e: logger.info(f"[ERROR] Failed parsing auth and exec id: {e}") + extra_info += f"\n{e}" # BASE URL (backend) try: @@ -3401,6 +3414,7 @@ class AppBase: logger.info(f"BACKEND URL: {app.url}") except Exception as e: logger.info(f"[ERROR] Failed parsing url (backend): {e}") + extra_info += f"\n{e}" # URL (worker) try: @@ -3408,6 +3422,7 @@ class AppBase: logger.info(f"WORKER URL: {app.base_url}") except Exception as e: logger.info(f"[ERROR] Failed parsing base url (worker): {e}") + extra_info += f"\n{e}" #await app.execute_action(app.action) @@ -3416,11 +3431,13 @@ class AppBase: return { "success": False, "reason": f"Problem in execution {e}", + "execution_issues": extra_info, } return { "success": True, "reason": "App successfully finished", + "execution_issues": extra_info, } else: return { @@ -3430,23 +3447,23 @@ class AppBase: logger.info(f"[DEBUG] Serving on port {port}") - flask_app.run( - host="0.0.0.0", - port=port, - threaded=True, - processes=1, - debug=False, - ) - - #serve( - # flask_app, + #flask_app.run( # host="0.0.0.0", # port=port, - # threads=8, - # channel_timeout=30, - # expose_tracebacks=True, - # asyncore_use_poll=True, + # threaded=True, + # processes=1, + # debug=False, #) + + serve( + flask_app, + host="0.0.0.0", + port=port, + threads=8, + channel_timeout=30, + expose_tracebacks=True, + asyncore_use_poll=True, + ) ####################### else: # Has to start like this due to imports in other apps diff --git a/backend/app_sdk/build.sh b/backend/app_sdk/build.sh index a1d5323f..8e92913b 100644 --- a/backend/app_sdk/build.sh +++ b/backend/app_sdk/build.sh @@ -3,21 +3,19 @@ ### DEFAULT NAME=shuffle-app_sdk -VERSION=0.9.66 +VERSION=0.9.67 docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force docker build . -f Dockerfile -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION -t ghcr.io/frikky/$NAME:nightly -#docker push frikky/$NAME:$VERSION -#docker push docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -#docker push ghcr.io/frikky/$NAME:$VERSION -#docker tag ghcr.io/frikky/$NAME:$VERSION frikky/shuffle:app_sdk - docker push frikky/shuffle:app_sdk docker push ghcr.io/frikky/$NAME:$VERSION docker push ghcr.io/frikky/$NAME:nightly docker push ghcr.io/frikky/$NAME:latest + + + #### KALI ### #NAME=shuffle-app_sdk_kali #docker build . -f Dockerfile_kali -t frikky/shuffle:app_sdk_kali -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION diff --git a/backend/go-app/docker.go b/backend/go-app/docker.go index 25acc65a..db57f05a 100644 --- a/backend/go-app/docker.go +++ b/backend/go-app/docker.go @@ -656,7 +656,7 @@ func getDockerImage(resp http.ResponseWriter, request *http.Request) { return } - log.Printf("[DEBUG] Image to load: %s", version.Name) + //log.Printf("[DEBUG] Image to load: %s", version.Name) dockercli, err := client.NewEnvClient() if err != nil { log.Printf("[WARNING] Unable to create docker client: %s", err) @@ -701,16 +701,16 @@ func getDockerImage(resp http.ResponseWriter, request *http.Request) { if len(img.ID) == 0 { if len(img2.ID) == 0 { workflowapps, err := shuffle.GetAllWorkflowApps(ctx, 0, 0) - log.Printf("[INFO] Getting workflowapps for a rebuild. Got %d with err %#v", len(workflowapps), err) + //log.Printf("[INFO] Getting workflowapps for a rebuild. Got %d with err %#v", len(workflowapps), err) if err == nil { imageName := "" imageVersion := "" newNameSplit := strings.Split(version.Name, ":") if len(newNameSplit) == 2 { - log.Printf("[DEBUG] Found name %#v", newNameSplit) + //log.Printf("[DEBUG] Found name %#v", newNameSplit) findVersionSplit := strings.Split(newNameSplit[1], "_") - log.Printf("[DEBUG] Found another split %#v", findVersionSplit) + //log.Printf("[DEBUG] Found another split %#v", findVersionSplit) if len(findVersionSplit) == 2 { imageVersion = findVersionSplit[len(findVersionSplit)-1] imageName = findVersionSplit[0] @@ -776,7 +776,7 @@ func getDockerImage(resp http.ResponseWriter, request *http.Request) { } //log.Printf("[INFO] Img found (%s): %#v", tagFound, img) - log.Printf("[INFO] Img found to be downloaded by client: %s", tagFound) + //log.Printf("[INFO] Img found to be downloaded by client: %s", tagFound) newClient, err := newdockerclient.NewClientFromEnv() if err != nil { diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index c6a9d450..19cd1694 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -24,7 +24,7 @@ require ( github.com/h2non/filetype v1.1.3 github.com/nirasan/go-oauth-pkce-code-verifier v0.0.0-20170819232839-0fbfe93532da // indirect github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.2.17 + github.com/shuffle/shuffle-shared v0.2.27 go4.org v0.0.0-20201209231011-d4a079459e60 // indirect golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce google.golang.org/api v0.65.0 diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 27ca1615..3e76b62c 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -376,10 +376,22 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) { // Authorization is done here if workflowExecution.Authorization != actionResult.Authorization { - log.Printf("[WARNING] Bad authorization key when getting stream results %s.", actionResult.ExecutionId) - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`))) - return + user, err := shuffle.HandleApiAuthentication(resp, request) + if err != nil { + log.Printf("[WARNING] Api authentication failed in exec grabbing workflow: %s", err) + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`))) + return + } + + if len(workflowExecution.ExecutionOrg) > 0 && user.ActiveOrg.Id == workflowExecution.ExecutionOrg && user.Role == "admin" { + log.Printf("[DEBUG] Correct org for execution!") + } else { + log.Printf("[WARNING] Bad authorization key when getting stream results %s.", actionResult.ExecutionId) + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`))) + return + } } newjson, err := json.Marshal(workflowExecution) @@ -828,7 +840,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request if workflow.ID == "" || workflow.ID != id { tmpworkflow, err := shuffle.GetWorkflow(ctx, id) if err != nil { - log.Printf("[WARNING] Failed getting the workflow locally (execution setup): %s", err) + //log.Printf("[WARNING] Failed getting the workflow locally (execution setup): %s", err) return shuffle.WorkflowExecution{}, "Failed getting workflow", err } diff --git a/docker-compose.yml b/docker-compose.yml index 1435d058..155eab06 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,7 +1,7 @@ version: '3' services: frontend: - #build: ./frontend + build: ./frontend image: ghcr.io/frikky/shuffle-frontend:nightly container_name: shuffle-frontend hostname: shuffle-frontend @@ -58,9 +58,9 @@ services: - HTTPS_PROXY=${HTTPS_PROXY} - SHUFFLE_PASS_WORKER_PROXY=${SHUFFLE_PASS_WORKER_PROXY} - SHUFFLE_PASS_APP_PROXY=${SHUFFLE_PASS_APP_PROXY} - - SHUFFLE_SWARM_NETWORK_NAME=shuffle_swarm_executions - SHUFFLE_SCALE_REPLICAS=1 - - SHUFFLE_SWARM_CONFIG=runn + - SHUFFLE_SWARM_CONFIG=run + - SHUFFLE_SWARM_NETWORK_NAME=shuffle_swarm_executions #- DOCKER_HOST=tcp://docker-socket-proxy:2375 restart: unless-stopped security_opt: @@ -118,6 +118,5 @@ services: # - shuffle networks: shuffle: - driver: bridge - -#driver: overlay + driver: overlay + #driver: bridge diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 44d64e19..70b74038 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -639,6 +639,7 @@ const AngularWorkflow = (defaultprops) => { typeof window === "undefined" || window.location === undefined ? "" : window.location.search; + var tmpView = new URLSearchParams(cursearch).get("execution_id"); if ( execution_id !== undefined && @@ -653,6 +654,7 @@ const AngularWorkflow = (defaultprops) => { const execution = responseJson.find( (data) => data.execution_id === tmpView ); + if (execution !== null && execution !== undefined) { setExecutionData(execution); setExecutionModalView(1); @@ -666,7 +668,25 @@ const AngularWorkflow = (defaultprops) => { const newitem = removeParam("execution_id", cursearch); navigate(curpath + newitem) //props.history.push(curpath + newitem); - } + } else { + console.log("Couldn't find execution for execution ID. Retrying as user to get ", tmpView) + + //setExecutionRequestStarted(true); + const cur_execution = { + execution_id: tmpView, + //authorization: data.authorization, + } + setExecutionModalView(1); + setExecutionRequest(cur_execution); + start(); + + const newitem = removeParam("execution_id", cursearch); + navigate(curpath + newitem) + + setTimeout(() => { + stop() + }, 5000); + } } } }) @@ -12248,7 +12268,7 @@ const parsedExecutionArgument = () => { }} onMouseOver={() => { var currentnode = cy.getElementById(data.action.id); - if (currentnode.length !== 0) { + if (currentnode !== undefined && currentnode !== null && currentnode.length !== 0) { currentnode.addClass("shuffle-hover-highlight"); } diff --git a/functions/onprem/orborus/build.sh b/functions/onprem/orborus/build.sh index 0f10630e..17303549 100644 --- a/functions/onprem/orborus/build.sh +++ b/functions/onprem/orborus/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-orborus -VERSION=0.9.65 +VERSION=0.9.67 echo "Running docker build with $NAME:$VERSION" #docker rmi frikky/shuffle:$NAME --force diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 1d686636..5b80eeb7 100644 --- a/functions/onprem/orborus/go.mod +++ b/functions/onprem/orborus/go.mod @@ -8,5 +8,5 @@ require ( github.com/docker/go-connections v0.4.0 // indirect github.com/mackerelio/go-osstat v0.2.1 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.2.9 + github.com/shuffle/shuffle-shared v0.2.27 ) diff --git a/functions/onprem/orborus/go.sum b/functions/onprem/orborus/go.sum index f75867e9..2b13181e 100644 --- a/functions/onprem/orborus/go.sum +++ b/functions/onprem/orborus/go.sum @@ -575,6 +575,8 @@ github.com/shuffle/shuffle-shared v0.1.73 h1:1rMOXAvxm/nDemwN/L8qWacswVMvdi6NjLf github.com/shuffle/shuffle-shared v0.1.73/go.mod h1:2ndjLm4ZOvY6arGFwOgGnkQ457Ke7gka9HDF/EkdIxQ= github.com/shuffle/shuffle-shared v0.2.9 h1:fh2eOD7olifW2uyC3Vlp8u3dqhgIxtYaDVjaYaNMXgA= github.com/shuffle/shuffle-shared v0.2.9/go.mod h1:YuMle0RjwXb3hxR5PdaOOD9e+hUyK34OABS0UbrT/Sk= +github.com/shuffle/shuffle-shared v0.2.27 h1:YT9MtXyMSxIGMpNovjp9pCKFyt2gk40EdAXqDvldhM8= +github.com/shuffle/shuffle-shared v0.2.27/go.mod h1:YuMle0RjwXb3hxR5PdaOOD9e+hUyK34OABS0UbrT/Sk= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc= github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc= diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 622927bc..5e21f808 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -273,7 +273,7 @@ func deployServiceWorkers(image string) { networkConfig := &network.EndpointSettings{} err := dockercli.NetworkConnect(ctx, networkName, containerId, networkConfig) if err != nil { - log.Printf("[ERROR] Failed connecting to Orborus to docker network %s: %s", networkName, err) + log.Printf("[ERROR] Failed connecting Orborus to docker network %s: %s", networkName, err) } if len(containerId) == 64 && baseUrl == "http://shuffle-backend:5001" { @@ -1358,6 +1358,6 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { _ = body - log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s.\n\n DEBUGGING: docker service logs shuffle-workers | grep %s\n\n", workflowExecution.ExecutionId, streamUrl, workflowExecution.ExecutionId) + log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s. DEBUGGING: docker service logs shuffle-workers | grep %s", workflowExecution.ExecutionId, streamUrl, workflowExecution.ExecutionId) return nil } diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 90ae079c..70ddb61e 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-worker -VERSION=0.9.66 +VERSION=0.9.67 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 633ab833..cc584910 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -10,6 +10,6 @@ require ( github.com/docker/go-connections v0.4.0 // indirect github.com/gorilla/mux v1.8.0 github.com/patrickmn/go-cache v2.1.0+incompatible - github.com/shuffle/shuffle-shared v0.2.22 + github.com/shuffle/shuffle-shared v0.2.27 go4.org v0.0.0-20201209231011-d4a079459e60 // indirect ) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index f483ef99..fe104e7e 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -912,7 +912,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { // get action status actionResult := getResult(workflowExecution, nextAction) if actionResult.Action.ID == action.ID { - log.Printf("[INFO] %s already has status %s.", action.ID, actionResult.Status) + //log.Printf("[INFO] %s already has status %s.", action.ID, actionResult.Status) continue } else { @@ -925,7 +925,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, nextAction) _, err := shuffle.GetCache(ctx, newExecId) if err == nil { - log.Printf("\n\n[DEBUG] Already found %s (1) - returning\n\n", newExecId) + //log.Printf("\n\n[DEBUG] Already found %s (1) - returning\n\n", newExecId) continue } @@ -1359,8 +1359,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } else { err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action) - log.Printf("[DEBUG] Failed deploying app? %s", err) if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { + log.Printf("[DEBUG] Failed deploying app? %s", err) if strings.Contains(err.Error(), "exited prematurely") { log.Printf("[DEBUG] Shutting down (9)") shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true) @@ -2074,7 +2074,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { if workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" { if workflowExecution.Workflow.Configuration.ExitOnError { - log.Printf("Workflowexecution already has status %s. No further action can be taken", workflowExecution.Status) + log.Printf("[WARNING] Workflowexecution already has status %s. No further action can be taken", workflowExecution.Status) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Workflowexecution is aborted because of %s with result %s and status %s"}`, workflowExecution.LastNode, workflowExecution.Result, workflowExecution.Status))) return @@ -2922,7 +2922,7 @@ func sendAppRequest(incomingUrl, appName string, port int, action shuffle.Action if err != nil { log.Printf("[WARNING] Failed setting cache for action %s: %s", newExecId, err) } else { - log.Printf("[DEBUG] Adding %s to cache. Name: %s", newExecId, action.Name) + log.Printf("[DEBUG] Adding %s to cache (%s)", newExecId, action.Name) } // FIXME: @@ -3035,7 +3035,8 @@ func baseDeploy() { //deployApp(cli, value, identifier, env, workflowExecution, action) log.Printf("[DEBUG] Deploying app with identifier %s to ensure basic apps are available from the get-go", identifier) - deployApp(cli, value, identifier, env, workflowExecution, action) + err = deployApp(cli, value, identifier, env, workflowExecution, action) + _ = err //err := deployApp(cli, value, identifier, env, workflowExecution, action) //if err != nil { // log.Printf("[DEBUG] Failed deploying app %s: %s", value, err)