marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed further execution concurrency issues with sdk & worker

Browse Source
This commit is contained in:
frikky
2022-04-07 21:44:42 +02:00
parent 20b483a1a6
commit b66d81178a
14 changed files with 108 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app_sdk/app_base.py
+39 -22
View File
@@ -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
backend/app_sdk/build.sh
+4 -6
View File
@@ -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
backend/go-app/docker.go
+5 -5
View File
@@ -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 {
backend/go-app/go.mod
+1 -1
View File
@@ -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
backend/go-app/walkoff.go
+17 -5
View File
@@ -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
}
docker-compose.yml
+5 -6
View File
@@ -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
frontend/src/views/AngularWorkflow.jsx
+22 -2
View File
@@ -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");
}
functions/onprem/orborus/build.sh
+1 -1
View File
@@ -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
functions/onprem/orborus/go.mod
+1 -1
View File
@@ -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
)
functions/onprem/orborus/go.sum
+2
View File
@@ -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=
functions/onprem/orborus/orborus.go
+2 -2
View File
@@ -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
}
functions/onprem/worker/build.sh
+1 -1
View File
@@ -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 .
functions/onprem/worker/go.mod
+1 -1
View File
@@ -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
)
functions/onprem/worker/worker.go
+7 -6
View File
@@ -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)