Merge branch 'nightly' of github.com:Shuffle/Shuffle into nightly

functions/kubernetes/charts/shuffle/templates/orborus/orborus-cm-env.yaml

@@ -26,23 +26,23 @@ data:
   {{- end }}
 
   # Shuffle worker resources
-  {{- $workerResources := (.Values.worker.resources | default (include "common.resources.preset" (dict "type" .Values.worker.resourcesPreset)) | fromYaml) -}}
-  {{- if $workerResources.requests.cpu }}
+  {{- $workerResources := (.Values.worker.resources | default (include "common.resources.preset" (dict "type" .Values.worker.resourcesPreset) | fromYaml)) -}}
+  {{- if and $workerResources.requests $workerResources.requests.cpu }}
   SHUFFLE_WORKER_CPU_REQUEST: {{ $workerResources.requests.cpu | quote }}
   {{- end }}
-  {{- if $workerResources.requests.memory}}
+  {{- if and $workerResources.requests $workerResources.requests.memory}}
   SHUFFLE_WORKER_MEMORY_REQUEST: {{ $workerResources.requests.memory | quote }}
   {{- end }}
-  {{- if (index $workerResources.requests "ephemeral-storage") }}
+  {{- if and $workerResources.requests (index $workerResources.requests "ephemeral-storage") }}
   SHUFFLE_WORKER_EPHEMERAL_STORAGE_REQUEST: {{ (index $workerResources.requests "ephemeral-storage") | quote }}
   {{- end }}
-  {{- if $workerResources.limits.cpu }}
+  {{- if and $workerResources.limits $workerResources.limits.cpu }}
   SHUFFLE_WORKER_CPU_LIMIT: {{ $workerResources.limits.cpu | quote }}
   {{- end }}
-  {{- if $workerResources.limits.memory}}
+  {{- if and $workerResources.limits $workerResources.limits.memory}}
   SHUFFLE_WORKER_MEMORY_LIMIT: {{ $workerResources.limits.memory | quote }}
   {{- end }}
-  {{- if (index $workerResources.limits "ephemeral-storage") }}
+  {{- if and $workerResources.limits (index $workerResources.limits "ephemeral-storage") }}
   SHUFFLE_WORKER_EPHEMERAL_STORAGE_LIMIT: {{ (index $workerResources.limits "ephemeral-storage") | quote }}
   {{- end }}
 
@@ -57,22 +57,22 @@ data:
   {{- end }}
 
   # Shuffle app resources
-  {{- $appResources := (.Values.app.resources | default (include "common.resources.preset" (dict "type" .Values.app.resourcesPreset)) | fromYaml) -}}
-  {{- if $appResources.requests.cpu }}
+  {{- $appResources := (.Values.app.resources | default (include "common.resources.preset" (dict "type" .Values.app.resourcesPreset) | fromYaml)) -}}
+  {{- if and $appResources.requests $appResources.requests.cpu }}
   SHUFFLE_APP_CPU_REQUEST: {{ $appResources.requests.cpu | quote }}
   {{- end }}
-  {{- if $appResources.requests.memory}}
+  {{- if and $appResources.requests $appResources.requests.memory }}
   SHUFFLE_APP_MEMORY_REQUEST: {{ $appResources.requests.memory | quote }}
   {{- end }}
-  {{- if (index $appResources.requests "ephemeral-storage") }}
+  {{- if and $appResources.requests (index $appResources.requests "ephemeral-storage") }}
   SHUFFLE_APP_EPHEMERAL_STORAGE_REQUEST: {{ (index $appResources.requests "ephemeral-storage") | quote }}
   {{- end }}
-  {{- if $appResources.limits.cpu }}
+  {{- if and $appResources.limits $appResources.limits.cpu }}
   SHUFFLE_APP_CPU_LIMIT: {{ $appResources.limits.cpu | quote }}
   {{- end }}
-  {{- if $appResources.limits.memory}}
+  {{- if and $appResources.limits $appResources.limits.memory }}
   SHUFFLE_APP_MEMORY_LIMIT: {{ $appResources.limits.memory | quote }}
   {{- end }}
-  {{- if (index $appResources.limits "ephemeral-storage") }}
+  {{- if and $appResources.limits (index $appResources.limits "ephemeral-storage") }}
   SHUFFLE_APP_EPHEMERAL_STORAGE_LIMIT: {{ (index $appResources.limits "ephemeral-storage") | quote }}
   {{- end }}

functions/onprem/orborus/go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/docker/docker v28.3.3+incompatible
 	github.com/docker/go-connections v0.5.0
 	github.com/satori/go.uuid v1.2.0
-	github.com/shuffle/shuffle-shared v0.9.27
+	github.com/shuffle/shuffle-shared v0.9.30
 	k8s.io/api v0.33.1
 	k8s.io/apimachinery v0.33.1
 )

functions/onprem/orborus/go.sum

@@ -328,8 +328,8 @@ github.com/sendgrid/sendgrid-go v3.16.1+incompatible h1:zWhTmB0Y8XCDzeWIm2/BIt1G
 github.com/sendgrid/sendgrid-go v3.16.1+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8=
 github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
 github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
-github.com/shuffle/shuffle-shared v0.9.27 h1:YwyWXsp4fCOAPmc1DD+NNf9sVa4RHzp26SvWKxH4ytc=
-github.com/shuffle/shuffle-shared v0.9.27/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw=
+github.com/shuffle/shuffle-shared v0.9.30 h1:3CYvNyD7sTxdxoZjTVrtaDqFvSWQRKAFGaga6rPGf8A=
+github.com/shuffle/shuffle-shared v0.9.30/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=

functions/onprem/orborus/orborus.go

@@ -1148,10 +1148,18 @@ func deployK8sWorker(image string, identifier string, env []string) error {
 		env = append(env, fmt.Sprintf("SHUFFLE_APP_CONTAINER_SECURITY_CONTEXT=%s", appContainerSecurityContext))
 	}
 
+	if len(os.Getenv("SHUFFLE_APP_MOUNT_TMP_VOLUME")) > 0 {
+		env = append(env, fmt.Sprintf("SHUFFLE_APP_MOUNT_TMP_VOLUME=%s", os.Getenv("SHUFFLE_APP_MOUNT_TMP_VOLUME")))
+	}
+
 	if len(os.Getenv("SHUFFLE_LOGS_DISABLED")) > 0 {
 		env = append(env, fmt.Sprintf("SHUFFLE_LOGS_DISABLED=%s", os.Getenv("SHUFFLE_LOGS_DISABLED")))
 	}
 
+	if len(os.Getenv("SHUFFLE_APP_REPLICAS")) > 0 {
+		env = append(env, fmt.Sprintf("SHUFFLE_APP_REPLICAS=%s", os.Getenv("SHUFFLE_APP_REPLICAS")))
+	}
+
 	clientset, _, err := shuffle.GetKubernetesClient()
 	if err != nil {
 		log.Printf("[ERROR] Error getting kubernetes client:", err)
@@ -1397,6 +1405,7 @@ func deployK8sWorker(image string, identifier string, env []string) error {
 		return err
 	}
 
+	svcAppProtocol := "http"
 	service := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   identifier,
@@ -1406,9 +1415,10 @@ func deployK8sWorker(image string, identifier string, env []string) error {
 			Selector: matchLabels,
 			Ports: []corev1.ServicePort{
 				{
-					Protocol:   "TCP",
-					Port:       33333,
-					TargetPort: intstr.FromInt(33333),
+					Protocol:    "TCP",
+					AppProtocol: &svcAppProtocol,
+					Port:        33333,
+					TargetPort:  intstr.FromInt(33333),
 				},
 			},
 			Type: corev1.ServiceTypeClusterIP,
@@ -2067,7 +2077,10 @@ func sendRemoveRequest(client *http.Client, toBeRemoved shuffle.ExecutionRequest
 
 	resultResp, err := client.Do(result)
 	if err != nil {
-		log.Printf("[ERROR] Failed making confirm request: %s", err)
+		if !strings.Contains(fmt.Sprintf("%s", err), "timeout") {
+			log.Printf("[ERROR] Failed making confirm request: %s", err)
+		}
+
 		time.Sleep(time.Duration(sleepTime) * time.Second)
 		return err
 	}
@@ -2196,12 +2209,23 @@ func main() {
 		os.Setenv("SHUFFLE_PIPELINE_ENABLED", "true")
 	}
 
-	log.Println("[INFO] Setting up execution environment")
+	if os.Getenv("SHUFFLE_SKIP_PIPELINES") != "true" && os.Getenv("SHUFFLE_PIPELINE_ENABLED") != "false" {
+		// Run in 15 seconds in a goroutine
+		go func() {
+			time.Sleep(15 * time.Second)
+			log.Printf("[INFO] Auto-downloading Sigma rules during startup")
+			ruleType := "sigma"
+			err := handleFileCategoryChange(ruleType)
+			if err != nil {
+				log.Printf("[WARNING] Failed downloading %s rules: %s", ruleType, err)
+			}
+		}()
+	}
 
+	log.Println("[INFO] Setting up execution environment for env '%s'", environment)
 	// //FIXME
 	if baseUrl == "" {
 		baseUrl = "https://shuffler.io"
-		//baseUrl = "http://localhost:5001"
 	}
 
 	if len(orborusUuid) == 0 {
@@ -2592,7 +2616,7 @@ func main() {
 					os.Setenv("SHUFFLE_SKIP_PIPELINES", "false")
 
 					tenzirDisabled = false
-					err = handleFileCategoryChange()
+					err = handleFileCategoryChange("sigma")
 					if err != nil {
 						log.Printf("[ERROR] Failed to download the file category: %s", err)
 					}
@@ -2602,7 +2626,7 @@ func main() {
 				} else if incRequest.Type == "DISABLE_SIGMA_FOLDER" {
 					log.Printf("[INFO] Got job to disable sigma rules")
 
-					err = removeFileCategory()
+					err = removeFileCategory("sigma")
 					if err != nil {
 						log.Printf("[ERROR] Failed to disable the sigma rules: %s", err)
 					}
@@ -3107,8 +3131,8 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 		Healthcheck: healthconfig,
 		ExposedPorts: nat.PortSet{
 			"5160/tcp": struct{}{},
-			"1514/udp":  struct{}{},
-			"1514/tcp":  struct{}{},
+			"1514/udp": struct{}{},
+			"1514/tcp": struct{}{},
 		},
 		Entrypoint: []string{containerName},
 		Env:        []string{},
@@ -3154,8 +3178,8 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 
 	hostConfig := &container.HostConfig{
 		PortBindings: nat.PortMap{
-			"1514/tcp":  []nat.PortBinding{{HostPort: "1514"}},
-			"1514/udp":  []nat.PortBinding{{HostPort: "1514"}},
+			"1514/tcp": []nat.PortBinding{{HostPort: "1514"}},
+			"1514/udp": []nat.PortBinding{{HostPort: "1514"}},
 			"5160/tcp": []nat.PortBinding{{HostPort: "5160"}},
 		},
 		Mounts: []mount.Mount{
@@ -3165,16 +3189,16 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 				Target: "/tmp",
 			},
 			/*
-			{
-				Type:   "bind",
-				Source: tenzirStorageFolder,
-				Target: "/var/log/tenzir/",
-			},
-			{
-				Type:   "bind",
-				Source: tenzirStorageFolder,
-				Target: "/var/cache/tenzir/",
-			},
+				{
+					Type:   "bind",
+					Source: tenzirStorageFolder,
+					Target: "/var/log/tenzir/",
+				},
+				{
+					Type:   "bind",
+					Source: tenzirStorageFolder,
+					Target: "/var/cache/tenzir/",
+				},
 			*/
 		},
 		VolumeDriver: "local",
@@ -3392,7 +3416,7 @@ func createPipeline(command, identifier string) (string, error) {
 		"name":        identifier,
 		"hidden":      false,
 		"retry_delay": "500.0ms",
-		"unstoppable":  true,
+		"unstoppable": true,
 	}
 
 	requestBodyJSON, err := json.Marshal(requestBody)
@@ -3464,20 +3488,20 @@ func updatePipelineState(command, pipelineId, action string) (string, error) {
 	url := fmt.Sprintf("%s/api/v0/pipeline/update", pipelineUrl)
 	forwardMethod := "POST"
 	requestBody := map[string]interface{}{
-		"id":         pipelineId,
-		"action":     action,
+		"id":     pipelineId,
+		"action": action,
 
 		/*
-		"autostart": map[string]bool{
-			"created":   true,
-			"completed": false,
-			"failed":    false,
-		},
-		"autodelete": map[string]bool{
-			"completed": false,
-			"failed":    false,
-			"stopped":   false,
-		},
+			"autostart": map[string]bool{
+				"created":   true,
+				"completed": false,
+				"failed":    false,
+			},
+			"autodelete": map[string]bool{
+				"completed": false,
+				"failed":    false,
+				"stopped":   false,
+			},
 		*/
 	}
 
@@ -3637,8 +3661,8 @@ func searchPipeline(identifier string) (string, error) {
 	return "", errors.New("no existing pipeline found with name")
 }
 
-func handleFileCategoryChange() error {
-	apiEndpoint := baseUrl + "/api/v1/files/namespaces/sigma"
+func handleFileCategoryChange(ruleType string) error {
+	apiEndpoint := fmt.Sprintf("%s/api/v1/files/namespaces/%s", baseUrl, ruleType)
 	req, err := http.NewRequest("GET", apiEndpoint, nil)
 	if err != nil {
 		return err
@@ -3685,14 +3709,13 @@ func handleFileCategoryChange() error {
 	}
 
 	//log.Println("[DEBUG] ZIP file downloaded successfully.")
-
 	tenzirStorageFolder := os.Getenv("SHUFFLE_STORAGE_FOLDER")
 	if len(tenzirStorageFolder) == 0 {
 		tenzirStorageFolder = "/tmp/"
 	}
 
 	tenzirStorageFolder = strings.TrimRight(tenzirStorageFolder, "/")
-	sigmaPath := fmt.Sprintf("%s/sigma_rules", tenzirStorageFolder)
+	sigmaPath := fmt.Sprintf("%s/%s_rules", tenzirStorageFolder, ruleType)
 	err = extractZIP("files.zip", sigmaPath)
 	if err != nil {
 		log.Printf("[ERROR] Failed to extract ZIP file: %s", err)
@@ -3780,7 +3803,7 @@ func copyToTenzir(srcPath, destPath string) error {
 	return nil
 }
 
-func removeFileCategory() error {
+func removeFileCategory(ruleType string) error {
 	tenzirStorageFolder := os.Getenv("SHUFFLE_STORAGE_FOLDER")
 	if len(tenzirStorageFolder) == 0 {
 		tenzirStorageFolder = "/tmp/"
@@ -3789,14 +3812,14 @@ func removeFileCategory() error {
 	tenzirStorageFolder = strings.TrimRight(tenzirStorageFolder, "/")
 
 	//sigmaPath := "/var/lib/tenzir/sigma_rules/*"
-	sigmaPath := fmt.Sprintf("%s/sigma_rules", tenzirStorageFolder)
+	rulePath := fmt.Sprintf("%s/%s_rules", tenzirStorageFolder, ruleType)
 
-	err := os.RemoveAll(sigmaPath)
+	err := os.RemoveAll(rulePath)
 	if err != nil {
-		return fmt.Errorf("Error removing category files in %s: %v", sigmaPath, err)
+		return fmt.Errorf("Error removing category files in %s: %v", rulePath, err)
 	}
 
-	log.Printf("[INFO] Removed all local category data in %s", sigmaPath)
+	log.Printf("[INFO] Removed all local category data in %s", rulePath)
 
 	return nil
 }
@@ -4216,7 +4239,9 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
 	newresp, err := client.Do(req)
 	if err != nil {
 		// Connection refused?
-		log.Printf("[ERROR][%s] Error running worker request to %s (1): %s", workflowExecution.ExecutionId, streamUrl, err)
+		if !strings.Contains(fmt.Sprintf("%s", err), "timeout") {
+			log.Printf("[ERROR][%s] Error running worker request to %s (1): %s", workflowExecution.ExecutionId, streamUrl, err)
+		}
 
 		if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
 			workerImage := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)

functions/onprem/worker/go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/docker/docker v28.3.3+incompatible
 	github.com/gorilla/mux v1.8.1
 	github.com/satori/go.uuid v1.2.0
-	github.com/shuffle/shuffle-shared v0.9.15
+	github.com/shuffle/shuffle-shared v0.9.30
 	github.com/shuffle/singul v0.0.16
 	k8s.io/api v0.33.1
 	k8s.io/apimachinery v0.33.1

functions/onprem/worker/worker.go

@@ -2,7 +2,7 @@ package main
 
 import (
 	"github.com/shuffle/shuffle-shared"
-	"github.com/shuffle/singul/pkg"
+	singul "github.com/shuffle/singul/pkg"
 
 	"bytes"
 	"context"
@@ -611,9 +611,9 @@ func deployk8sApp(image string, identifier string, env []string) error {
 
 	// use deployment instead of pod
 	// then expose a service similarly.
-	// number of replicas can be set to os.Getenv("SHUFFLE_SCALE_REPLICAS")
+	// number of replicas can be set to os.Getenv("SHUFFLE_APP_REPLICAS")
 	replicaNumber := 1
-	replicaNumberStr := os.Getenv("SHUFFLE_SCALE_REPLICAS")
+	replicaNumberStr := os.Getenv("SHUFFLE_APP_REPLICAS")
 	if len(replicaNumberStr) > 0 {
 		tmpInt, err := strconv.Atoi(replicaNumberStr)
 		if err != nil {
@@ -664,6 +664,27 @@ func deployk8sApp(image string, identifier string, env []string) error {
 		},
 	}
 
+	if os.Getenv("SHUFFLE_APP_MOUNT_TMP_VOLUME") == "true" {
+		deployment.Spec.Template.Spec.Volumes = append(
+			deployment.Spec.Template.Spec.Volumes,
+			corev1.Volume{
+				Name: "tmp",
+				VolumeSource: corev1.VolumeSource{
+					EmptyDir: &corev1.EmptyDirVolumeSource{},
+				},
+			},
+		)
+
+		deployment.Spec.Template.Spec.Containers[0].VolumeMounts = append(
+			deployment.Spec.Template.Spec.Containers[0].VolumeMounts,
+			corev1.VolumeMount{
+				Name:      "tmp",
+				ReadOnly:  false,
+				MountPath: "/tmp",
+			},
+		)
+	}
+
 	if len(os.Getenv("REGISTRY_URL")) > 0 && len(os.Getenv("SHUFFLE_BASE_IMAGE_NAME")) > 0 {
 		log.Printf("[INFO] Setting image pull policy to Always as private registry is used.")
 		//containerAttachment.ImagePullPolicy = corev1.PullAlways
@@ -678,6 +699,7 @@ func deployk8sApp(image string, identifier string, env []string) error {
 		return err
 	}
 
+	svcAppProtocol := "http"
 	service := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   name,
@@ -687,9 +709,10 @@ func deployk8sApp(image string, identifier string, env []string) error {
 			Selector: matchLabels,
 			Ports: []corev1.ServicePort{
 				{
-					Protocol:   "TCP",
-					Port:       80,
-					TargetPort: intstr.FromInt(deployport),
+					Protocol:    "TCP",
+					AppProtocol: &svcAppProtocol,
+					Port:        80,
+					TargetPort:  intstr.FromInt(deployport),
 				},
 			},
 			Type: corev1.ServiceTypeClusterIP,
@@ -2093,6 +2116,11 @@ func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.Workflow
 		}
 	}
 
+	if len(data) == 0 {
+		log.Printf("[WARNING] Stream result missing execution ID and authorization; injecting them from workflow execution")
+		data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, workflowExecution.ExecutionId, workflowExecution.Authorization)
+	}
+
 	req, err := http.NewRequest(
 		"POST",
 		streamResultUrl,
@@ -3233,8 +3261,6 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl
 	log.Printf("[DEBUG] Deploying service for %s to swarm on port %d", name, deployport)
 	//containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid)
 
-
-
 	// Check if the image exists or not - just in case
 	_, _, err := dockercli.ImageInspectWithRaw(context.Background(), image)
 	if err != nil {
@@ -3247,8 +3273,8 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl
 		}
 
 		_, err := dockercli.ImagePull(
-			context.Background(), 
-			image, 
+			context.Background(),
+			image,
 			dockerimage.PullOptions{},
 		)
 		if err != nil {
@@ -3257,7 +3283,6 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl
 		}
 	}
 
-
 	if len(baseimagename) == 0 || baseimagename == "/" {
 		baseimagename = "frikky/shuffle"
 		//var baseimagename = "frikky/shuffle"
@@ -3290,7 +3315,7 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl
 
 	// Max scale as well
 	nodeCount := uint64(1)
-	if inputReplicas > 0 && inputReplicas < 100 { 
+	if inputReplicas > 0 && inputReplicas < 100 {
 		if replicas != uint64(inputReplicas) {
 			log.Printf("[DEBUG] Overwriting replicas to %d/node as inputReplicas is set to %d", inputReplicas, inputReplicas)
 		}
@@ -3310,7 +3335,6 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl
 		nodeCount = 1
 	}
 
-
 	replicatedJobs := uint64(replicas * nodeCount)
 	log.Printf("[DEBUG] Deploying app with name %s with image %s", name, image)
 
@@ -3440,12 +3464,12 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl
 			}
 
 			// Retry deploying the service (once)
-			if !retry { 
+			if !retry {
 				return deploySwarmService(dockercli, name, image, deployport, -1, true)
 			}
 		}
 
-		// For port mapping. 
+		// For port mapping.
 		if strings.Contains(fmt.Sprintf("%s", err), "InvalidArgument") && strings.Contains(fmt.Sprintf("%s", err), "is already in use") {
 			//log.Printf("\n\n[WARNING] Port %d is already allocated. Trying to deploy on next port.\n\n", deployport)
 
@@ -3472,7 +3496,6 @@ func findAppInfo(image, name string, redeploy bool) (int, error) {
 	// chance of being successful
 	time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
 
-
 	highest := baseport
 	exposedPort := -1
 
@@ -3576,10 +3599,10 @@ func findAppInfo(image, name string, redeploy bool) (int, error) {
 						time.Sleep(time.Duration(rand.Intn(4)+8) * time.Second)
 						replicas := service.Spec.Mode.Replicated.Replicas
 						err = deploySwarmService(
-							dockercli, 
-							name, 
-							image, 
-							exposedPort, 
+							dockercli,
+							name,
+							image,
+							exposedPort,
 							int64(*replicas),
 							false,
 						)
@@ -3669,7 +3692,7 @@ func findAppInfoKubernetes(image, name string, env []string) error {
 
 	for _, deployment := range deployments.Items {
 		if deployment.Name == name {
-			if debug { 
+			if debug {
 				log.Printf("[DEBUG] Found deployment %s - no need to deploy another", name)
 			}
 
@@ -3695,7 +3718,6 @@ func initSwarmNetwork() error {
 	mtu := 1500
 	options["com.docker.network.driver.mtu"] = fmt.Sprintf("%d", mtu)
 
-
 	ingressOptions := network.CreateOptions{
 		Driver:     "overlay",
 		Attachable: false,
@@ -3779,11 +3801,9 @@ func initSwarmNetwork() error {
 		log.Printf("[WARNING] Swarm Executions network may already exist: %s", err)
 	}
 
-	return nil 
+	return nil
 }
 
-
-
 /*** ENDREMOVE ***/
 
 func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution, image string, attempts int64) error {
@@ -3940,9 +3960,9 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int,
 		}
 
 		// Try redeployment
-		attempts += 1 
+		attempts += 1
 		if attempts < 2 {
-			// Check the service and fix it. 
+			// Check the service and fix it.
 			if isKubernetes == "true" {
 				log.Printf("[WARNING] App Redeployment in K8s isn't fully supported yet, but should be done for app %s with image %s.", appName, image)
 			} else {
@@ -3951,7 +3971,7 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int,
 					log.Printf("[ERROR][%s] Error re-deploying app %s: %s", workflowExecution.ExecutionId, appName, err)
 				}
 
-				return sendAppRequest(ctx, incomingUrl, appName, port, action, workflowExecution, image, attempts) 
+				return sendAppRequest(ctx, incomingUrl, appName, port, action, workflowExecution, image, attempts)
 			}
 		}
 
@@ -4377,7 +4397,7 @@ func checkStandaloneRun() {
 
 // Initial loop etc
 func main() {
-	// Testing swarm auto-replacements. This also tests ports 
+	// Testing swarm auto-replacements. This also tests ports
 	// in rapid succession
 
 	checkStandaloneRun()
@@ -4385,7 +4405,7 @@ func main() {
 		debug = true
 
 		log.Printf("[INFO] Disabled cleanup due to debug mode (DEBUG=true)")
-		cleanupEnv = "false" 
+		cleanupEnv = "false"
 	}
 
 	/*** STARTREMOVE ***/
@@ -4835,7 +4855,7 @@ func runWebserver(listener net.Listener) {
 	if strings.ToLower(os.Getenv("SHUFFLE_SWARM_CONFIG")) == "run" || strings.ToLower(os.Getenv("SHUFFLE_APP_REPLICAS")) == "" {
 		// go AutoScaleApps(ctx, dockercli, maxExecutionsPerMinute)
 	}
-	if (strings.ToLower(os.Getenv("SHUFFLE_DEBUG_MEMORY")) == "true" || strings.ToLower(os.Getenv("DEBUG_MEMORY")) == "true") {
+	if strings.ToLower(os.Getenv("SHUFFLE_DEBUG_MEMORY")) == "true" || strings.ToLower(os.Getenv("DEBUG_MEMORY")) == "true" {
 		r.HandleFunc("/debug/pprof/", pprof.Index)
 		r.HandleFunc("/debug/pprof/heap", pprof.Handler("heap").ServeHTTP)
 		r.HandleFunc("/debug/pprof/profile", pprof.Profile)