Merge branch '2.0.0' of https://github.com/shuffle/shuffle into 2.0.0

.github/workflows/dockerbuild.yaml

@@ -27,11 +27,11 @@ jobs:
             experimental: true
           - app: orborus
             path: functions/onprem/orborus
-            version: nightly
+            version: latest
             experimental: true
           - app: worker
             path: functions/onprem/worker
-            version: nightly
+            version: latest
             experimental: true
     steps:
       - name: Checkout

functions/onprem/orborus/orborus.go

@@ -333,19 +333,6 @@ func deployServiceWorkers(image string) {
 	}
 	ctx := context.Background()
 
-	isMemcachedRunning, err := checkMemcached(ctx, dockercli)
-	if err != nil {
-		log.Printf("[ERROR] Failed checking memcached: %s", err)
-	}
-	if isMemcachedRunning == false {
-		log.Printf("[ERROR] Memcached is not running. Will try to deploy it.")
-		deployMemcached(dockercli)
-	}
-
-	ip := "shuffle-cache"
-
-	os.Setenv("SHUFFLE_MEMCACHED", fmt.Sprintf("%s:11211", ip))
-
 	// Looks for and cleans up all existing items in swarm we can't re-use (Shuffle only)
 
 	// frikky@debian:~/git/shuffle/functions/onprem/worker$ docker service create --replicas 5 --name shuffle-workers --env SHUFFLE_SWARM_CONFIG=run --publish published=33333,target=33333 ghcr.io/shuffle/shuffle-worker:nightly
@@ -457,6 +444,19 @@ func deployServiceWorkers(image string) {
 		}
 	}
 
+	isMemcachedRunning, err := checkMemcached(ctx, dockercli)
+	if err != nil {
+		log.Printf("[ERROR] Failed checking memcached: %s", err)
+	}
+	if isMemcachedRunning == false {
+		log.Printf("[ERROR] Memcached is not running. Will try to deploy it.")
+		deployMemcached(dockercli)
+	}
+
+	ip := "shuffle-cache"
+
+	os.Setenv("SHUFFLE_MEMCACHED", fmt.Sprintf("%s:11211", ip))
+
 	defaultNetworkAttach := false
 	if containerId != "" {
 		log.Printf("[DEBUG] Should connect orborus container to worker network as it's running in Docker with name %#v!", containerId)
@@ -2150,7 +2150,9 @@ func main() {
 				log.Printf("[DEBUG] Starting iteration on environment %#v (default = Shuffle). Got statuscode %d from backend on first request", environment, newresp.StatusCode)
 			}
 
-			go AutoScale(ctx)
+			if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" && os.Getenv("SHUFFLE_SCALE_REPLICAS") == "" {
+				go AutoScale(ctx)
+			}
 			hasStarted = true
 		}
 
@@ -2624,7 +2626,6 @@ func deployTenzirNode() error {
 
 	err := checkTenzirNode()
 	if err == nil {
-		log.Printf("[INFO] Tenzir Node is already running")
 		return nil
 	}
 
@@ -2930,9 +2931,7 @@ func checkTenzirNode() error {
 		return nil
 	}
 
-	log.Printf("[DEBUG] Failed to verify Tenzir node on %s: %s", url, err)
+	return fmt.Errorf("Tenzir node is not available due to: %s", err)
-
-	return fmt.Errorf("Tenzir node is not available")
 }
 
 func createPipeline(command, identifier string) (string, error) {
@@ -3961,6 +3960,11 @@ func checkMemcached(ctx context.Context, dockercli *dockerclient.Client) (bool,
 		}
 		return false, err
 	}
+	networkName := "shuffle_swarm_executions"
+	err = dockercli.NetworkConnect(ctx, networkName, containerName, nil)
+	if err != nil {
+		log.Printf("[WARNING] Failed connecting memcached container to network: %s", err)
+	}
 
 	if continer.State.Running == false {
 		log.Printf("[INFO] Container %s exists but is not running. Attempting to start it.", containerName)
@@ -3998,19 +4002,43 @@ func deployMemcached(dockercli *dockerclient.Client) error {
 		},
 	}
 
-	dockercli.ImagePull(ctx, memcachedImage, image.PullOptions{})
+	_, _, err := dockercli.ImageInspectWithRaw(ctx, memcachedImage)
-	containerName := "shuffle-cache"
+	if dockerclient.IsErrNotFound(err) {
+		log.Printf("[DEBUG] Pulling image %s. This may take a while.", memcachedImage)
+		pullOptions := image.PullOptions{}
+		out, err := dockercli.ImagePull(ctx, memcachedImage, pullOptions)
+		if err != nil {
+			log.Printf("[ERROR] Failed to pull the memcached image: %s", err)
+			return err
+		}
+		defer out.Close()
 
+		io.Copy(io.Discard, out)
+	} else if err != nil {
+		return err
+	}
+
+	containerName := "shuffle-cache"
 	resp, err := dockercli.ContainerCreate(ctx, containerConfig, hostConfig, nil, nil, containerName)
 	if err != nil {
 		log.Printf("[ERROR] Error spanning memcached continer: %s", err)
 		return err
 	}
+
+	if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" {
+		networkName := "shuffle_swarm_executions"
+		err = dockercli.NetworkConnect(ctx, networkName, resp.ID, nil)
+		if err != nil {
+			log.Printf("[ERROR] Error connecting tenzir container to network: %s", err)
+		}
+	}
+
 	err = dockercli.ContainerStart(ctx, resp.ID, container.StartOptions{})
 	if err != nil {
 		log.Printf("[ERROR] Error starting memcached continer: %s", err)
 		return err
 	}
+
 	networkName := "shuffle_swarm_executions"
 	err = dockercli.NetworkConnect(ctx, networkName, resp.ID, nil)
 	if err != nil {

functions/onprem/worker/worker.go

@@ -104,11 +104,11 @@ var window = shuffle.NewTimeWindow(10 * time.Second)
 
 // Images to be autodeployed in the latest version of Shuffle.
 var autoDeploy = map[string]string{
-	"http:1.4.0":            "frikky/shuffle:http_1.4.0",
+	"http:1.4.0":               "frikky/shuffle:http_1.4.0",
-	"http:1.3.0":            "frikky/shuffle:http_1.3.0",
+	"http:1.3.0":               "frikky/shuffle:http_1.3.0",
-	"shuffle-tools:1.2.0":   "frikky/shuffle:shuffle-tools_1.2.0",
+	"shuffle-tools:1.2.0":      "frikky/shuffle:shuffle-tools_1.2.0",
-	"shuffle-subflow:1.0.0": "frikky/shuffle:shuffle-subflow_1.0.0",
+	"shuffle-subflow:1.0.0":    "frikky/shuffle:shuffle-subflow_1.0.0",
-	"shuffle-subflow:1.1.0": "frikky/shuffle:shuffle-subflow_1.1.0",
+	"shuffle-subflow:1.1.0":    "frikky/shuffle:shuffle-subflow_1.1.0",
 	"shuffle-tools-fork:1.0.0": "frikky/shuffle:shuffle-tools-fork_1.0.0",
 }
 
@@ -4129,8 +4129,9 @@ func runWebserver(listener net.Listener) {
 		log.Printf("[DEBUG] SHUFFLE_APP_EXECUTIONS_PER_MINUTE set to value %s. Trying to overwrite default (%d)", os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE"), maxExecutionsPerMinute)
 	}
 
-	go AutoScaleApps(ctx, dockercli, maxExecutionsPerMinute)
+	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" {
 		r.HandleFunc("/debug/pprof/", pprof.Index)
 		r.HandleFunc("/debug/pprof/heap", pprof.Handler("heap").ServeHTTP)
@@ -4179,8 +4180,6 @@ func AutoScaleApps(ctx context.Context, client *dockerclient.Client, maxExecutio
 			j := numberOfApps(ctx, client)
 			workers := numberOfWorkers(ctx, client)
 			execPerMin := maxExecutionsPerMinute / workers
-			log.Printf("[DEBUG] Running with %d workers\n\n\n\n\n", workers)
-
 			if count >= execPerMin {
 				log.Printf("[DEBUG] Too many executions per minute (%d). Scaling down to %d", count, execPerMin)
 				scaleApps(ctx, client, uint64(j+1))