diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 60a01b11..adb69f94 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -507,7 +507,7 @@ func setWorkflowQueue(ctx context.Context, executionRequests ExecutionRequestWra } func getWorkflowQueue(ctx context.Context, id string) (ExecutionRequestWrapper, error) { - key := datastore.NameKey("workflowqueue", id, nil) + key := datastore.NameKey("workflowqueue", id, nil).Limit(50) workflows := ExecutionRequestWrapper{} if err := dbclient.Get(ctx, key, &workflows); err != nil { return ExecutionRequestWrapper{}, err diff --git a/backend/tests/execute.sh b/backend/tests/execute.sh index a4584bbe..dd8d7a16 100644 --- a/backend/tests/execute.sh +++ b/backend/tests/execute.sh @@ -1,5 +1,29 @@ #!/bin/sh -curl -XPOST http://localhost:5001/api/v1/workflows/1d9d8ce2-566e-4c3f-8a37-5d6c7d2000b5/execute -d '{"execution_argument":""}' -H "Authorization: Bearer 144308d0-6aab-4d4f-8bb2-75189281ee26" +#curl -XPOST http://localhost:5001/api/v1/workflows/1d9d8ce2-566e-4c3f-8a37-5d6c7d2000b5/execute -d '{"execution_argument":""}' -H "Authorization: Bearer 144308d0-6aab-4d4f-8bb2-75189281ee26" -curl -XPOST http://localhost:5001/api/v1/workflows/1d9d8ce2-566e-4c3f-8a37-5d6c7d2000b5/execute -d '{"execution_argument":""}' -H "Authorization: Bearer eyJhbGciOiJSUzI1NiIsImtpZCI6IjYwZjQwNjBlNThkNzVmZDNmNzBiZWZmODhjNzk0YTc3NTMyN2FhMzEiLCJ0eXAiOiJKV1QifQ.eyJhdWQiOiJodHRwczovL3NodWZmbGVyLmlvL2FwaS92MS93b3JrZmxvd3MvMWQ5ZDhjZTItNTY2ZS00YzNmLThhMzctNWQ2YzdkMjAwMGI1L2V4ZWN1dGUiLCJhenAiOiIxMDMwNzY3ODIwNjE0MjQ2MTg0MjIiLCJlbWFpbCI6InNjaGVkdWxlckBzaHVmZmxlLTI0MTUxNy5pYW0uZ3NlcnZpY2VhY2NvdW50LmNvbSIsImVtYWlsX3ZlcmlmaWVkIjp0cnVlLCJleHAiOjE1NjU1Mjc1NTEsImlhdCI6MTU2NTUyMzk1MSwiaXNzIjoiaHR0cHM6Ly9hY2NvdW50cy5nb29nbGUuY29tIiwic3ViIjoiMTAzMDc2NzgyMDYxNDI0NjE4NDIyIn0.r0EDq9fjhf_5CPTiltyfk_L3uYJp577Uy0yYPcCAl2nv50_z_oUtbWGBpQLL8gcj-NGd3g4E52Qur8k6hCMIQweLS6WAb1279vGffEoCNDfkWb3Oy-yJGP1kzwLvqFJqnHLkSWYXNWvSyWnEimW8Rryx_m1BXS5wcA8l4NIr83kS7fPZrTwjnwFSeGSThwk91DVARzapQb8r0GEgOUyHZ1aBXnV98mikzSUt-5xFKe9eMdD22YJAj0Ru-DxAxs5nOqghX4PMRysWjshjOMrlR1piPWxqAmewp8YKZDCQ5gXskpeAFBDoULT971Wsx_NCohnJsFqx1JfPS9ZYMTW2oQ" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" diff --git a/docker-compose.yml b/docker-compose.yml index 81934cf1..8e0565a0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,8 +1,8 @@ version: '3' services: frontend: - #build: ./frontend - image: ghcr.io/frikky/shuffle-frontend:0.8.43 + build: ./frontend + image: ghcr.io/frikky/shuffle-frontend:0.8.45 container_name: shuffle-frontend hostname: shuffle-frontend ports: @@ -17,7 +17,7 @@ services: - backend backend: #build: ./backend - image: ghcr.io/frikky/shuffle-backend:0.8.44 + image: ghcr.io/frikky/shuffle-backend:0.8.45 container_name: shuffle-backend hostname: ${BACKEND_HOSTNAME} # Here for debugging: diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index a2adc14d..4a3a2439 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -3681,7 +3681,7 @@ const AngularWorkflow = (props) => {

{selectedAction.app_name}

-
+
{ console.log("FIND EXAMPLE RESULTS FOR ", selectedAction) if (workflowExecutions.length > 0) { @@ -3707,7 +3707,7 @@ const AngularWorkflow = (props) => { - + What are actions? {selectedAction.errors !== null && selectedAction.errors.length > 0 ?
@@ -6294,17 +6294,6 @@ const AngularWorkflow = (props) => { style={{pointerEvents: "none"}} hideBackdrop={true} open={codeModalOpen} - onClose={() => { - //setCodeModalOpen(false) - console.log("CLOSE?") - }} - BackdropProps={{ - invisible: true, - style: { - backgroundColor: "transparent", - pointerEvents: "none", - } - }} PaperProps={{ style: { pointerEvents: "auto", @@ -6313,7 +6302,8 @@ const AngularWorkflow = (props) => { minWidth: 750, padding: 30, maxHeight: 700, - overflow: "auto", + overflowY: "auto", + overflowX: "hidden", //boxShadow: "none", }, }} @@ -6828,7 +6818,7 @@ const AngularWorkflow = (props) => { id="copy_element_shuffle" value={to_be_copied} disabled={true} - style={{height: 0, width: 0, margin: 0, padding: 0,}} + style={{height: 0, width: 0, margin: 0, padding: 0, zIndex: -10, position: "fixed",}} />
: diff --git a/functions/onprem/orborus/build.sh b/functions/onprem/orborus/build.sh index 57fe63a2..2403b361 100644 --- a/functions/onprem/orborus/build.sh +++ b/functions/onprem/orborus/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-orborus -VERSION=0.8.32 +VERSION=0.8.40 echo "Running docker build with $NAME:$VERSION" #docker rmi frikky/shuffle:$NAME --force diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index cf6a8b1a..39b8e3a3 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -29,9 +29,11 @@ import ( // Starts jobs in bulk, so this could be increased var sleepTime = 3 +var maxConcurrency = 50 // Timeout if something rashes var workerTimeoutEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_TIMEOUT") +var concurrencyEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY") var appSdkVersion = os.Getenv("SHUFFLE_APP_SDK_VERSION") var workerVersion = os.Getenv("SHUFFLE_WORKER_VERSION") @@ -147,7 +149,7 @@ func deployWorker(image string, identifier string, env []string) { Env: env, } - log.Printf("[INFO] Identifier: %s", identifier) + //log.Printf("[INFO] Identifier: %s", identifier) cont, err := dockercli.ContainerCreate( context.Background(), config, @@ -317,7 +319,18 @@ func main() { log.Printf("[INFO] Cleanup process running every %d seconds", workerTimeout) } - go zombiecheck(workerTimeout) + if concurrencyEnv != "" { + //var concurrencyEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY") + tmpInt, err := strconv.Atoi(concurrencyEnv) + if err == nil { + maxConcurrency = tmpInt + } else { + log.Printf("[WARNING] Env SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY must be a number, not %s. Defaulted to %d", workerTimeoutEnv, maxConcurrency) + } + } + + ctx := context.Background() + go zombiecheck(ctx, workerTimeout) log.Printf("[INFO] Running towards %s with Org %s", baseUrl, orgId) httpProxy := os.Getenv("HTTP_PROXY") @@ -383,12 +396,13 @@ func main() { for { //log.Printf("Prerequest") newresp, err := client.Do(req) + executionCount := getRunningWorkers(ctx, workerTimeout) //log.Printf("Postrequest") if err != nil { log.Printf("[WARNING] Failed making request: %s", err) zombiecounter += 1 if zombiecounter*sleepTime > workerTimeout { - go zombiecheck(workerTimeout) + go zombiecheck(ctx, workerTimeout) zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) @@ -409,7 +423,7 @@ func main() { log.Printf("[ERROR] Failed reading body: %s", err) zombiecounter += 1 if zombiecounter*sleepTime > workerTimeout { - go zombiecheck(workerTimeout) + go zombiecheck(ctx, workerTimeout) zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) @@ -423,7 +437,7 @@ func main() { sleepTime = 10 zombiecounter += 1 if zombiecounter*sleepTime > workerTimeout { - go zombiecheck(workerTimeout) + go zombiecheck(ctx, workerTimeout) zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) @@ -438,13 +452,31 @@ func main() { if len(executionRequests.Data) == 0 { zombiecounter += 1 if zombiecounter*sleepTime > workerTimeout { - go zombiecheck(workerTimeout) + go zombiecheck(ctx, workerTimeout) zombiecounter = 0 } time.Sleep(time.Duration(sleepTime) * time.Second) continue } + // Anything below here verifies concurrency virification + if executionCount >= maxConcurrency { + if zombiecounter*sleepTime > workerTimeout { + go zombiecheck(ctx, workerTimeout) + zombiecounter = 0 + } + time.Sleep(time.Duration(sleepTime) * time.Second) + continue + } + + log.Printf("Got %d new requests. Executing: %d. Max: %d", len(executionRequests.Data), executionCount, maxConcurrency) + + allowed := maxConcurrency - executionCount + if len(executionRequests.Data) > allowed { + log.Printf("[WARNING] Throttle - Cutting down requests from %d to %d", len(executionRequests.Data), allowed) + executionRequests.Data = executionRequests.Data[0:allowed] + } + // New, abortable version. Should check executionid and remove everything else var toBeRemoved ExecutionRequestWrapper for _, execution := range executionRequests.Data { @@ -481,7 +513,7 @@ func main() { go deployWorker(workerImage, containerName, env) - log.Printf("[INFO] %s is deployed and to be removed from queue.", execution.ExecutionId) + log.Printf("[INFO] %s was deployed and to be removed from queue.", execution.ExecutionId) zombiecounter += 1 toBeRemoved.Data = append(toBeRemoved.Data, execution) } @@ -543,12 +575,59 @@ func main() { } } +// Is this ok to do with Docker? idk :) +func getRunningWorkers(ctx context.Context, workerTimeout int) int { + containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{ + All: true, + }) + + if err != nil { + log.Printf("Error getting containers: %s", err) + return 0 + } + + currenttime := time.Now().Unix() + counter := 0 + for _, container := range containers { + // Skip random containers. Only handle things related to Shuffle. + if !strings.Contains(container.Image, baseimagename) { + shuffleFound := false + for _, item := range container.Labels { + if item == "shuffle" { + shuffleFound = true + break + } + } + + // Check image name + if !shuffleFound { + continue + } + //} else { + // log.Printf("NAME: %s", container.Image) + } + + for _, name := range container.Names { + // FIXME - add name_version_uid_uid regex check as well + if !strings.HasPrefix(name, "/worker") { + continue + } + + //log.Printf("Time: %d - %d", currenttime-container.Created, int64(workerTimeout)) + if container.State == "running" && currenttime-container.Created < int64(workerTimeout) { + counter += 1 + break + } + } + } + + return counter +} + // FIXME - add this to remove exited workers // Should it check what happened to the execution? idk -func zombiecheck(workerTimeout int) error { +func zombiecheck(ctx context.Context, workerTimeout int) error { log.Println("[INFO] Looking for old containers") - ctx := context.Background() - containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{ All: true, })