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

Fixed async issues with apps everywhere

Browse Source
This commit is contained in:
frikky
2021-11-01 15:05:56 +01:00
parent 502ed68091
commit 3185a157e5
8 changed files with 3816 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app_sdk/app_base.py
+79 -23
View File
@@ -9,6 +9,7 @@ import logging
import urllib3
import hashlib
import zipfile
import asyncio
import requests
import http.client
import urllib.parse
@@ -113,9 +114,11 @@ class AppBase:
self.logger.info(f"[DEBUG] Expected ProtocolError happened: {e}")
return
async def cartesian_product(self, L):
#async def cartesian_product(self, L):
def cartesian_product(self, L):
if L:
return {(a, ) + b for a in L[0] for b in await self.cartesian_product(L[1:])}
#return {(a, ) + b for a in L[0] for b in await self.cartesian_product(L[1:])}
return {(a, ) + b for a in L[0] for b in self.cartesian_product(L[1:])}
else:
return {()}
@@ -251,7 +254,8 @@ class AppBase:
# Returns a list of all the executions to be done in the inner loop
# FIXME: Doesn't take into account whether you actually WANT to loop or not
# Check if the last part of the value is #?
async def get_param_multipliers(self, baseparams):
#async def get_param_multipliers(self, baseparams):
def get_param_multipliers(self, baseparams):
# Example:
# {'call': ['hello', 'hello4'], 'call2': ['hello2', 'hello3'], 'call3': '1'}
#
@@ -429,7 +433,8 @@ class AppBase:
self.logger.info("[DEBUG] Newlength of array: %d. Lists: %s" % (newlength, all_lists))
# Get the cartesian product of the arrays
cartesian = await self.cartesian_product(all_lists)
#cartesian = await self.cartesian_product(all_lists)
cartesian = self.cartesian_product(all_lists)
newlist = []
for item in cartesian:
newlist.append(list(item))
@@ -458,7 +463,8 @@ class AppBase:
# Runs recursed versions with inner loops and such
async def run_recursed_items(self, func, baseparams, loop_wrapper):
#async def run_recursed_items(self, func, baseparams, loop_wrapper):
def run_recursed_items(self, func, baseparams, loop_wrapper):
#self.logger.info(f"RECURSED ITEMS: {baseparams}")
has_loop = False
@@ -500,13 +506,15 @@ class AppBase:
results = []
if has_loop:
self.logger.info(f"[DEBUG] Should run inner loop: {newparams}")
ret = await self.run_recursed_items(func, newparams, loop_wrapper)
#ret = await self.run_recursed_items(func, newparams, loop_wrapper)
ret = self.run_recursed_items(func, newparams, loop_wrapper)
else:
self.logger.info(f"[DEBUG] Should run multiplier check with params (inner): {newparams}")
# 1. Find the loops that are required and create new multipliers
# If here: check for multipliers within this scope.
ret = []
param_multiplier = await self.get_param_multipliers(newparams)
#param_multiplier = await self.get_param_multipliers(newparams)
param_multiplier = self.get_param_multipliers(newparams)
# FIXME: This does a deduplication of the data
new_params = self.validate_unique_fields(param_multiplier)
@@ -553,7 +561,8 @@ class AppBase:
while True:
try:
tmp = await func(**subparams)
#tmp = await func(**subparams)
tmp = func(**subparams)
break
except TypeError as e:
self.logger.info("BASE TYPEERROR: %s" % e)
@@ -581,6 +590,25 @@ class AppBase:
tmp = "An error occured during execution: %s" % e
# An attempt at decomposing coroutine results
try:
if asyncio.iscoroutine(tmp):
print("In coroutine")
async def parse_value(tmp):
value = await asyncio.gather(
tmp
)
return value[0]
tmp = asyncio.run(parse_value(tmp))
else:
print("Not in coroutine")
except Exception as e:
print("[ERROR] Failed to parse coroutine value for old app: {e}")
#self.logger.info("RET from execution: %s" % ret)
new_value = tmp
if tmp == None:
@@ -2124,7 +2152,7 @@ class AppBase:
try:
func = getattr(self, actionname, None)
if func == None:
self.logger.debug(f"[DEBUG] Failed executing {actionname} because func is None.")
self.logger.debug(f"[DEBUG] Failed executing {actionname} because func is None (no function specified).")
self.action_result["status"] = "FAILURE"
self.action_result["result"] = "Function %s doesn't exist." % actionname
elif callable(func):
@@ -2584,6 +2612,23 @@ class AppBase:
raise e
#break
# Forcing async wait in case of old apps that use async
try:
if asyncio.iscoroutine(newres):
print("In coroutine")
async def parse_value(newres):
value = await asyncio.gather(
newres
)
return value[0]
newres = asyncio.run(parse_value(newres))
else:
print("Not in coroutine")
except Exception as e:
print("[ERROR] Failed to parse coroutine value for old app: {e}")
self.logger.info("\n[INFO] Returned from execution with types %s" % type(newres))
#self.logger.info("\n[INFO] Returned from execution with %s of types %s" % (newres, type(newres)))#, newres)
if isinstance(newres, tuple):
@@ -2830,7 +2875,7 @@ class AppBase:
return
@classmethod
async def run(cls, action=""):
def run(cls, action=""):
logging.basicConfig(format="{asctime} - {name} - {levelname}:{message}", style='{')
logger = logging.getLogger(f"{cls.__name__}")
logger.setLevel(logging.DEBUG)
@@ -2846,12 +2891,15 @@ class AppBase:
logger.info(f"[DEBUG] Starting webserver on port {port} (same as exposed port)")
from flask import Flask, request
from waitress import serve
import asyncio
flask_app = Flask(__name__)
@flask_app.route("/api/v1/run", methods=["POST"])
#async def execute():
@flask_app.route("/api/v1/health", methods=["GET", "POST"])
def check_health():
return "OK"
@flask_app.route("/api/v1/run", methods=["POST"])
def execute():
if request.method == "POST":
#print(request.get_json(force=True))
@@ -2894,18 +2942,18 @@ class AppBase:
app.url = requestdata["url"]
logger.info(f"BACKEND URL: {app.url}")
except:
logger.info("Failed parsing url")
logger.info("Failed parsing url (backend)")
# URL (worker)
try:
app.base_url = requestdata["base_url"]
logger.info(f"WORKER URL: {app.base_url}")
except:
logger.info("Failed parsing base url")
logger.info("Failed parsing base url (worker)")
#await
app.execute_action(app.action)
logger.info("\n\n[DEBUG] Done awaiting app action running\n\n")
logger.info("[DEBUG] Done awaiting app action running")
except Exception as e:
return {
"success": False,
@@ -2923,15 +2971,23 @@ class AppBase:
}
logger.info(f"[DEBUG] Serving on port {port}")
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
@@ -2971,5 +3027,5 @@ class AppBase:
#app.run(host="0.0.0.0", port=33334)
if __name__ == "__main__":
import asyncio
asyncio.run(AppBase.run(), debug=True)
AppBase.run()
#asyncio.run(AppBase.run(), debug=True)
backend/go-app/go.mod
+1 -1
View File
@@ -21,7 +21,7 @@ require (
github.com/gorilla/mux v1.8.0
github.com/h2non/filetype v1.1.1
github.com/satori/go.uuid v1.2.0
github.com/shuffle/shuffle-shared v0.1.27
github.com/shuffle/shuffle-shared v0.1.28
go4.org v0.0.0-20201209231011-d4a079459e60 // indirect
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
google.golang.org/api v0.58.0
frontend/src/views/Workflows.jsx
+8 -1
View File
@@ -278,7 +278,14 @@ export const validateJson = (showResult) => {
}
}
const result = jsonvalid ? JSON.parse(showResult) : showResult
var result = showResult
try {
const result = jsonvalid ? JSON.parse(showResult) : showResult
} catch (e) {
//console.log("Failed parsing JSON even though its valid: ", e)
jsonvalid = false
}
//console.log("VALID: ", jsonvalid, result)
return {
"valid": jsonvalid,
functions/onprem/orborus/go.sum
+1084
View File
File diff suppressed because it is too large Load Diff
functions/onprem/orborus/orborus.go
+34 -25
View File
@@ -236,7 +236,7 @@ func deployServiceWorkers(image string) {
}
if len(os.Getenv("SHUFFLE_SCALE_REPLICAS")) > 0 {
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("DOCKER_API_VERSION=%s", os.Getenv("SHUFFLE_SCALE_REPLICAS")))
serviceSpec.TaskTemplate.ContainerSpec.Env = append(serviceSpec.TaskTemplate.ContainerSpec.Env, fmt.Sprintf("SHUFFLE_SCALE_REPLICAS=%s", os.Getenv("SHUFFLE_SCALE_REPLICAS")))
}
serviceOptions := types.ServiceCreateOptions{}
@@ -301,10 +301,15 @@ func deployWorker(image string, identifier string, env []string, executionReques
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
} else {
}
if err == nil {
log.Printf("[DEBUG] Started worker from request with name: %s", executionRequest.ExecutionId)
executionIds = append(executionIds, executionRequest.ExecutionId)
}
}()
@@ -347,7 +352,6 @@ func deployWorker(image string, identifier string, env []string, executionReques
containerStartOptions := types.ContainerStartOptions{}
err = dockercli.ContainerStart(context.Background(), cont.ID, containerStartOptions)
if err != nil {
log.Printf("[DEBUG] Failed initial container start. Running WITHOUT custom network. Err: %s", err)
// Trying to recreate and start WITHOUT network if it's possible. No extended checks. Old execution system (<0.9.30)
if strings.Contains(fmt.Sprintf("%s", err), "cannot join network") || strings.Contains(fmt.Sprintf("%s", err), "No such container") {
hostConfig.NetworkMode = ""
@@ -362,6 +366,8 @@ func deployWorker(image string, identifier string, env []string, executionReques
)
err = dockercli.ContainerStart(context.Background(), cont.ID, containerStartOptions)
} else {
log.Printf("[ERROR] Failed initial container start. Quitting as this is NOT a simple network issue. Err: %s", err)
}
if err != nil {
@@ -677,31 +683,34 @@ func main() {
// Type string `json:"type"`
}
if len(executionRequests.Data) == 0 {
zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
zombiecounter = 0
// Skipping throttling with swarm
if swarmConfig != "run" {
if len(executionRequests.Data) == 0 {
zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
zombiecounter = 0
}
time.Sleep(time.Duration(sleepTime) * time.Second)
continue
}
time.Sleep(time.Duration(sleepTime) * time.Second)
continue
}
// Anything below here verifies concurrency
executionCount := getRunningWorkers(ctx, workerTimeout)
if executionCount >= maxConcurrency {
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
zombiecounter = 0
// Anything below here verifies concurrency
executionCount := getRunningWorkers(ctx, workerTimeout)
if executionCount >= maxConcurrency {
if zombiecounter*sleepTime > workerTimeout {
go zombiecheck(ctx, workerTimeout)
zombiecounter = 0
}
time.Sleep(time.Duration(sleepTime) * time.Second)
continue
}
time.Sleep(time.Duration(sleepTime) * time.Second)
continue
}
allowed := maxConcurrency - executionCount
if len(executionRequests.Data) > allowed {
log.Printf("[WARNING] Throttle - Cutting down requests from %d to %d (MAX: %d, CUR: %d)", len(executionRequests.Data), allowed, maxConcurrency, executionCount)
executionRequests.Data = executionRequests.Data[0:allowed]
allowed := maxConcurrency - executionCount
if len(executionRequests.Data) > allowed {
log.Printf("[WARNING] Throttle - Cutting down requests from %d to %d (MAX: %d, CUR: %d)", len(executionRequests.Data), allowed, maxConcurrency, executionCount)
executionRequests.Data = executionRequests.Data[0:allowed]
}
}
// New, abortable version. Should check executionid and remove everything else
functions/onprem/worker/go.sum
+1085
View File
File diff suppressed because it is too large Load Diff
functions/onprem/worker/run.sh
+1514 -1
View File
File diff suppressed because it is too large Load Diff
functions/onprem/worker/worker.go
+11 -4
View File
@@ -54,6 +54,8 @@ var topClient *http.Client
var data string
var requestsSent = 0
var hostname string
/*
var environments []string
var parents map[string][]string
@@ -1919,7 +1921,7 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) {
//if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions)+extra {
if (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1) || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions) && len(workflowExecution.Workflow.Actions) > 0) {
requestsSent += 1
log.Printf("[FINISHED] Should send full result to %s", baseUrl)
log.Printf("[DEBUG] Should send full result to %s", baseUrl)
//data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization)
shutdownData, err := json.Marshal(workflowExecution)
@@ -2039,7 +2041,7 @@ func getAvailablePort() (net.Listener, error) {
}
func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
hostname := getLocalIP()
hostname = getLocalIP()
// FIXME: This MAY not work because of speed between first
// container being launched and port being assigned to webserver
@@ -2406,6 +2408,11 @@ func sendAppRequest(incomingUrl string, port int, action shuffle.Action, workflo
parsedRequest.Url = parsedRequest.BaseUrl
parsedRequest.BaseUrl = tmp
if len(hostname) > 0 {
log.Printf("[DEBUG] Changing hostname to local hostname in Docker network for WORKER URL")
parsedRequest.BaseUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
}
log.Printf("[DEBUG] Worker URL: %s, Backend URL: %s", parsedRequest.BaseUrl, parsedRequest.Url)
data, err := json.Marshal(parsedRequest)
@@ -2423,13 +2430,13 @@ func sendAppRequest(incomingUrl string, port int, action shuffle.Action, workflo
client := &http.Client{}
if err != nil {
log.Printf("[ERROR] Failed creating finishing request: %s", err)
log.Printf("[ERROR] Failed creating app run request: %s", err)
return err
}
newresp, err := client.Do(req)
if err != nil {
log.Printf("[ERROR] Error running finishing request: %s", err)
log.Printf("[ERROR] Error running app run request: %s", err)
return err
}