Updated worker nightly to run with latest shared 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.5.31
+	github.com/shuffle/shuffle-shared v0.5.53
 	k8s.io/api v0.28.3
 	k8s.io/apimachinery v0.28.3
 	k8s.io/client-go v0.28.3

functions/onprem/worker/go.sum

@@ -288,6 +288,10 @@ github.com/shuffle/shuffle-shared v0.5.29 h1:n4vThl7v3mFVXbrIW71XREFdmZZo7mOBAWx
 github.com/shuffle/shuffle-shared v0.5.29/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI=
 github.com/shuffle/shuffle-shared v0.5.31 h1:OV4IIfKWWFW66WjGvyXOmmsSz3p8pW9L1ge1mDo8ftM=
 github.com/shuffle/shuffle-shared v0.5.31/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI=
+github.com/shuffle/shuffle-shared v0.5.44 h1:6WiFPIsij+IWvXY7vzVX7cUicb+PYOzhTbWF/gDmYeU=
+github.com/shuffle/shuffle-shared v0.5.44/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI=
+github.com/shuffle/shuffle-shared v0.5.53 h1:Osr5sjr0wppqlf/MjH1oykdcw/Q48d8+ZnerRrYfOA4=
+github.com/shuffle/shuffle-shared v0.5.53/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI=
 github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e h1:MRM5ITcdelLK2j1vwZ3Je0FKVCfqOLp5zO6trqMLYs0=
 github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e/go.mod h1:XV66xRDqSt+GTGFMVlhk3ULuV0y9ZmzeVGR4mloJI3M=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=

functions/onprem/worker/worker.go

@@ -354,7 +354,8 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
 		req.Header.Add("Content-Type", "application/json")
 
 		//log.Printf("[DEBUG][%s] All App Logs: %#v", workflowExecution.ExecutionId, allLogs)
-		newresp, err := topClient.Do(req)
+		client := shuffle.GetExternalClient(abortUrl)
+		newresp, err := client.Do(req)
 		if err != nil {
 			log.Printf("[WARNING][%s] Failed abort request: %s", workflowExecution.ExecutionId, err)
 		} else {
@@ -464,6 +465,9 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
 	// Check action if subflow
 	// Check if url is default (shuffle-backend)
 	// If it doesn't exist, add it
+
+	// FIXME: This does NOT replace it in all cases as the data
+	// is not saved in the database as the correct param.
 	if action.AppName == "shuffle-subflow" {
 		// Automatic replacement of URL
 		for paramIndex, param := range action.Parameters {
@@ -471,7 +475,10 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
 				continue
 			}
 
-			if strings.Contains(param.Value, "shuffle-backend") {
+			if !strings.Contains(param.Value, "shuffle-backend") {
+				continue
+			} 
+
 			// Automatic replacement as this is default
 			if len(os.Getenv("BASE_URL")) > 0 {
 				action.Parameters[paramIndex].Value = os.Getenv("BASE_URL")
@@ -484,7 +491,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
 			}
 		}
 	}
-	}
 
 
 	// Max 10% CPU every second
@@ -745,6 +751,8 @@ func removeContainer(containername string) error {
 		return err
 	}
 
+	defer cli.Close()
+
 	// FIXME - ucnomment
 	//	containers, err := cli.ContainerList(ctx, types.ContainerListOptions{
 	//		All: true,
@@ -800,6 +808,8 @@ func getWorkerURLs() ([]string, error) {
 		return workerUrls, err
 	}
 
+	defer cli.Close()
+
 	// Specify the name of the service for which you want to list tasks
 	serviceName := "shuffle-workers"
 
@@ -827,12 +837,19 @@ func askOtherWorkersToDownloadImage(image string) {
 		return
 	}
 
+	// Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD
+	if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" {
+		log.Printf("[DEBUG] SHUFFLE_AUTO_IMAGE_DOWNLOAD is false. NOT distributing images %s", image)
+		return 
+	}
+
 	urls, err := getWorkerURLs()
 	if err != nil {
 		log.Printf("[ERROR] Error in listing worker urls: %s", err)
 		return
 	}
 
+	httpClient := &http.Client{}
 	for _, url := range urls {
 		log.Printf("[DEBUG] Trying to speak to: %s", url)
 		imagesRequest := ImageRequest{
@@ -855,7 +872,6 @@ func askOtherWorkersToDownloadImage(image string) {
 			continue
 		}
 
-		httpClient := &http.Client{}
 		resp, err := httpClient.Do(req)
 		if err != nil {
 			log.Printf("[ERROR] Error in making request to %s : %s", url, err)
@@ -892,6 +908,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
 		return
 	}
 
+	defer dockercli.Close()
+
 	for _, action := range relevantActions {
 		appname := action.AppName
 		appversion := action.AppVersion
@@ -1116,6 +1134,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
 							shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
 							return
 						} else {
+							defer reader.Close()
 							baseTag := strings.Split(image, ":")
 							if len(baseTag) > 1 {
 								tag := baseTag[1]
@@ -1226,6 +1245,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
 								shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
 								return
 							} else {
+								defer reader.Close()
 								baseTag := strings.Split(image, ":")
 								if len(baseTag) > 1 {
 									tag := baseTag[1]
@@ -1494,7 +1514,8 @@ func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.Workflow
 		bytes.NewBuffer([]byte(data)),
 	)
 
-	newresp, err := topClient.Do(req)
+	client := shuffle.GetExternalClient(streamResultUrl)
+	newresp, err := client.Do(req)
 	if err != nil {
 		log.Printf("[ERROR] Failed making request (1): %s", err)
 		time.Sleep(time.Duration(sleepTime) * time.Second)
@@ -2101,7 +2122,8 @@ func sendSelfRequest(actionResult shuffle.ActionResult) {
 		return
 	}
 
-	newresp, err := topClient.Do(req)
+	client := shuffle.GetExternalClient(streamUrl)
+	newresp, err := client.Do(req)
 	if err != nil {
 		log.Printf("[ERROR][%s] Error running finishing request (2): %s", actionResult.ExecutionId, err)
 		return
@@ -2158,9 +2180,10 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) {
 		return
 	}
 
-	newresp, err := topClient.Do(req)
+	client := shuffle.GetExternalClient(streamUrl)
+	newresp, err := client.Do(req)
 	if err != nil {
-		log.Printf("[ERROR][%s] Error running finishing request: %s", workflowExecution.ExecutionId, err)
+		log.Printf("[ERROR][%s] Error running finishing request (1): %s", workflowExecution.ExecutionId, err)
 		log.Printf("[DEBUG][%s] Shutting down (23)", workflowExecution.ExecutionId)
 		shutdown(workflowExecution, "", "", false)
 		return
@@ -2348,6 +2371,12 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
 func downloadDockerImageBackend(client *http.Client, imageName string) error {
 	log.Printf("[DEBUG] Trying to download image %s from backend %s as it doesn't exist. All images: %#v", imageName, baseUrl, downloadedImages)
 
+	// Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD
+	if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" {
+		log.Printf("[DEBUG] SHUFFLE_AUTO_IMAGE_DOWNLOAD is false. Not downloading image %s", imageName)
+		return nil
+	}
+
 	if arrayContains(downloadedImages, imageName) {
 		log.Printf("[DEBUG] Image %s already downloaded", imageName)
 		return nil
@@ -2407,12 +2436,15 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error {
 		return err
 	}
 
+	defer dockercli.Close()
+
 	imageLoadResponse, err := dockercli.ImageLoad(context.Background(), tar, true)
 	if err != nil {
 		log.Printf("[ERROR] Error loading images: %s", err)
 		return err
 	}
 
+	defer imageLoadResponse.Body.Close()
 	body, err := ioutil.ReadAll(imageLoadResponse.Body)
 	if err != nil {
 		log.Printf("[ERROR] Error reading: %s", err)
@@ -2570,9 +2602,11 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int,
 		log.Printf("[DEBUG][%s] Adding %s to cache (%#v)", workflowExecution.ExecutionId, newExecId, action.Name)
 	}
 
-	// FIXME: Add 5 tries 
+	client := shuffle.GetExternalClient(streamUrl)
 
-	newresp, err := topClient.Do(req)
+	// Set client timeout to 5 seconds
+	client.Timeout = time.Duration(10) * time.Second
+	newresp, err := client.Do(req)
 	if err != nil {
 		// Another timeout issue here somewhere
 		// context deadline
@@ -2630,6 +2664,8 @@ func baseDeploy() {
 		return
 	}
 
+	defer cli.Close()
+
 	for key, value := range autoDeploy {
 		newNameSplit := strings.Split(key, ":")
 
@@ -2897,7 +2933,6 @@ func main() {
 		shutdown(workflowExecution, "", "", true)
 	}
 
-	topClient = client
 	firstRequest := true
 	environments := []string{}
 	for {
@@ -3007,15 +3042,14 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
 	var workflowExecution shuffle.WorkflowExecution
 	data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, execRequest.ExecutionId, execRequest.Authorization)
 	streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl)
-	topClient = shuffle.GetExternalClient(streamResultUrl)
-
 	req, err := http.NewRequest(
 		"POST",
 		streamResultUrl,
 		bytes.NewBuffer([]byte(data)),
 	)
 
-	newresp, err := topClient.Do(req)
+	client := shuffle.GetExternalClient(streamResultUrl)
+	newresp, err := client.Do(req)
 	if err != nil {
 		log.Printf("[ERROR] Failed making request (2): %s", err)
 		resp.WriteHeader(401)
@@ -3145,6 +3179,8 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) {
 		return
 	}
 
+	defer client.Close()
+
 	// check if images are already downloaded
 	// Retrieve a list of Docker images
 	images, err := client.ImageList(context.Background(), types.ImageListOptions{})
@@ -3155,6 +3191,7 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) {
 		return
 	}
 
+
 	for _, img := range images {
 		for _, tag := range img.RepoTags {
 			splitTag := strings.Split(tag, ":")