From c6f32f6ddf3e79a4c96c62510799f5503cffcfd4 Mon Sep 17 00:00:00 2001 From: frikky Date: Thu, 18 May 2023 15:31:00 +0200 Subject: [PATCH] Cache related backend fixes --- backend/go-app/go.mod | 2 +- backend/go-app/walkoff.go | 10 +++++----- docker-compose.yml | 1 - functions/onprem/orborus/orborus.go | 2 +- functions/onprem/worker/build.sh | 4 ++-- functions/onprem/worker/go.mod | 2 +- functions/onprem/worker/worker.go | 14 ++++++++------ 7 files changed, 18 insertions(+), 17 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index d1ae292e..7b2250b6 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -19,7 +19,7 @@ require ( github.com/gorilla/mux v1.8.0 github.com/h2non/filetype v1.1.3 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.4.11 + github.com/shuffle/shuffle-shared v0.4.13 golang.org/x/crypto v0.3.0 google.golang.org/api v0.103.0 google.golang.org/appengine v1.6.7 diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index a6bdb2a0..e903ae04 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -105,7 +105,7 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode } } - log.Printf("Starting frequency: %d", newfrequency) + log.Printf("[INFO] Starting frequency for execution: %d", newfrequency) jobret, err := newscheduler.Every(newfrequency).Seconds().NotImmediately().Run(job) if err != nil { log.Printf("Failed to schedule workflow: %s", err) @@ -333,7 +333,7 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { // Check if CPU percent constantly has stayed above X% for the last Y requests percentageCheck := 90 - concurrentChecks := 0 + concurrentChecks := 2 //if int(envData.CPUPercent) > percentageCheck { // Get cached data @@ -1801,15 +1801,15 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) { return } - log.Printf("Starting Cloud schedule Action: %#v", action) + //log.Printf("Starting Cloud schedule Action: %#v", action) err = executeCloudAction(action, org.SyncConfig.Apikey) if err != nil { - log.Printf("Failed cloud action START schedule: %s", err) + log.Printf("[WARNING] Failed cloud action START schedule: %s", err) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) return } else { - log.Printf("Successfully set up cloud action schedule") + log.Printf("[INFO] Successfully set up cloud action schedule") resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Done"}`))) return diff --git a/docker-compose.yml b/docker-compose.yml index 1143f01c..bd3e42a7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -43,7 +43,6 @@ services: - /var/run/docker.sock:/var/run/docker.sock environment: #- DOCKER_HOST=tcp://docker-socket-proxy:2375 - - SHUFFLE_WORKER_VERSION=latest - ENVIRONMENT_NAME=${ENVIRONMENT_NAME} - BASE_URL=http://${OUTER_HOSTNAME}:5001 - DOCKER_API_VERSION=1.40 diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index abec33ee..71e128ae 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -726,7 +726,7 @@ func initializeImages() { if baseimagename == "" { baseimagename = "shuffle/shuffle" // Dockerhub - baseimagename = "shuffle" // Github + baseimagename = "shuffle" // Github (ghcr.io) log.Printf("[DEBUG] Setting baseimagename") } diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index b12f5a7e..28d6a47b 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -1,9 +1,9 @@ NAME=shuffle-worker -VERSION=1.1.5 +VERSION=1.2.0 echo "Running docker build with $NAME:$VERSION" #CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin . -docker build . -t frikky/shuffle:$NAME -t frikky/shuffle:$NAME_$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION -t ghcr.io/frikky/$NAME:nightly +docker build . -t frikky/shuffle:$NAME -t frikky/shuffle:$NAME_$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION -t ghcr.io/frikky/$NAME:nightly -t ghcr.io/shuffle/$NAME:$VERSION -t ghcr.io/shuffle/$NAME:nightly # Push both for now.. #docker push frikky/$NAME:$VERSION diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index 8820d6b3..addc04cf 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -11,7 +11,7 @@ require ( github.com/gorilla/mux v1.8.0 github.com/patrickmn/go-cache v2.1.0+incompatible github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.4.11 + github.com/shuffle/shuffle-shared v0.4.12 ) require ( diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index bc72e7b8..7a1570a7 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -1637,7 +1637,7 @@ func sendSelfRequest(actionResult shuffle.ActionResult) { newresp, err := topClient.Do(req) if err != nil { - log.Printf("[ERROR][%s] Error running finishing request (2): %s", actionResult.ExecutionId, err) + log.Printf("[ERROR][%s] Error running self request (2): %s", actionResult.ExecutionId, err) return } @@ -1646,16 +1646,18 @@ func sendSelfRequest(actionResult shuffle.ActionResult) { body, err := ioutil.ReadAll(newresp.Body) //log.Printf("[INFO] BACKEND STATUS: %d", newresp.StatusCode) if err != nil { - log.Printf("[ERROR][%s] Failed reading body: %s", actionResult.ExecutionId, err) + log.Printf("[ERROR][%s] Failed reading self request body: %s", actionResult.ExecutionId, err) } else { - log.Printf("[DEBUG][%s] NEWRESP (from backend): %s", actionResult.ExecutionId, string(body)) + log.Printf("[DEBUG][%s] NEWRESP (from self - 1): %s", actionResult.ExecutionId, string(body)) } } } func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { - log.Printf("[INFO][%s] Not sending backend info since source is default (not swarm)", workflowExecution.ExecutionId) - return + if workflowExecution.ExecutionSource == "default" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" { + //log.Printf("[INFO][%s] Not sending backend info since source is default (not swarm)", workflowExecution.ExecutionId) + //return + } streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl) req, err := http.NewRequest( @@ -2003,7 +2005,7 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error { // Initial loop etc func main() { // Elasticsearch necessary to ensure we'ren ot running with Datastore configurations for minimal/maximal data sizes - _, err := shuffle.RunInit(datastore.Client{}, storage.Client{}, "", "", true, "elasticsearch") + _, err := shuffle.RunInit(datastore.Client{}, storage.Client{}, "", "worker", true, "elasticsearch") if err != nil { log.Printf("[ERROR] Failed to run worker init: %s", err) } else {