Cache related backend fixes

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

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

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

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")
 	}
 

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 

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 (

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)
+	if workflowExecution.ExecutionSource == "default" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
-	return
+		//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 {