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

backend/go-app/main.go

@@ -4009,22 +4009,32 @@ func runInitEs(ctx context.Context) {
 			log.Printf("[DEBUG] Creating org for default user %s", username)
 			orgId := uuid.NewV4().String()
 			orgSetupName := "default"
+			tmpOrg := shuffle.OrgMini{
+				Name: orgSetupName,
+				Id:   orgId,
+			}
+			err = createNewUser(username, password, "admin", apikey, tmpOrg)
+			if err != nil {
+				log.Printf("[ERROR] Failed to create default user %s: %s", username, err)
+			} else {
+				log.Printf("[INFO] Successfully created user %s", username)
+			}
+
+			user, err := shuffle.GetUser(ctx, username)
 			newOrg := shuffle.Org{
 				Name:      orgSetupName,
 				Id:        orgId,
 				Org:       orgSetupName,
-				Users:     []shuffle.User{},
+				Users:     []shuffle.User{*user},
 				Roles:     []string{"admin", "user"},
 				CloudSync: false,
 			}
 
 			err = shuffle.SetOrg(ctx, newOrg, newOrg.Id)
-			setUsers := false
 			if err != nil {
-				log.Printf("[WARNING] Failed setting organization when creating original user: %s", err)
+				log.Printf("[ERROR] Failed setting organization when creating original user: %s", err)
 			} else {
 				log.Printf("[DEBUG] Successfully created the default org with id %s!", orgId)
-				setUsers = true
 
 				item := shuffle.Environment{
 					Name:    defaultEnv,
@@ -4039,20 +4049,6 @@ func runInitEs(ctx context.Context) {
 					log.Printf("[WARNING] Failed setting up new environment")
 				}
 			}
-
-			if setUsers {
-				tmpOrg := shuffle.OrgMini{
-					Name: orgSetupName,
-					Id:   orgId,
-				}
-
-				err = createNewUser(username, password, "admin", apikey, tmpOrg)
-				if err != nil {
-					log.Printf("[INFO] Failed to create default user %s: %s", username, err)
-				} else {
-					log.Printf("[INFO] Successfully created user %s", username)
-				}
-			}
 		}
 	} else {
 		for _, user := range users {
@@ -5026,7 +5022,7 @@ func initHandlers() {
 		go runInitEs(ctx)
 	} else {
 		//go shuffle.runInit(ctx)
-		log.Printf("[ERROR] Opensearch is the only viable option. Please set SHUFFLE_ELASTIC=true") 
+		log.Printf("[ERROR] Opensearch is the only viable option. Please set SHUFFLE_ELASTIC=true")
 		os.Exit(1)
 	}
 

functions/onprem/orborus/orborus.go

@@ -15,7 +15,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/shuffle/shuffle-shared"
 	"io"
 	"io/ioutil"
 	"log"
@@ -31,6 +30,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/shuffle/shuffle-shared"
+
 	"math/rand"
 	//"os/signal"
 	//"syscall"
@@ -102,6 +103,8 @@ var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG")
 var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME")
 var orborusLabel = os.Getenv("SHUFFLE_ORBORUS_LABEL")
 var memcached = os.Getenv("SHUFFLE_MEMCACHED")
+var queuePerMinute = os.Getenv("SHUFFLE_EXECUTION_PER_MINIUTE")
+var queuePerMinuteInt int
 
 // For it to download from Sigma?
 var apiKey = os.Getenv("AUTH_FOR_ORBORUS")
@@ -110,14 +113,15 @@ var pipelineUrl = os.Getenv("SHUFFLE_PIPELINE_URL")
 var executionIds = []string{}
 var pipelines = []shuffle.PipelineInfoMini{}
 var namespacemade = false // For K8s
-var skipPipelineMount = false 
-var tenzirDisabled = false 
+var skipPipelineMount = false
+var tenzirDisabled = false
 
 var dockercli *dockerclient.Client
 var containerId string
 var executionCount = 0
 
 var imagedownloadTimeout = time.Second * 300
+var window = shuffle.NewTimeWindow(1 * time.Minute)
 
 func init() {
 	var err error
@@ -194,9 +198,7 @@ func skipCheckInCleanup(name string) bool {
 		strings.HasPrefix(name, "orborus") ||
 		strings.HasPrefix(name, "shuffle-orborus") ||
 		strings.HasPrefix(name, "opensearch") ||
-		strings.HasPrefix(name, "shuffle-opensearch") ||
-		strings.HasPrefix(name, "memcached") ||
-		strings.HasPrefix(name, "shuffle-memcached")
+		strings.HasPrefix(name, "shuffle-opensearch")
 }
 
 func cleanupExistingNodes(ctx context.Context) error {
@@ -291,7 +293,6 @@ func cleanupExistingNodes(ctx context.Context) error {
 	//log.Printf("\n\nFound %d contaienrs", len(services))
 
 	for _, service := range services {
-		//log.Printf("[INFO] Service: %#v", service.Spec.Annotations.Name)
 
 		//portFound := false
 		//for _, endpoint := range service.Spec.EndpointSpec.Ports {
@@ -329,9 +330,20 @@ func deployServiceWorkers(image string) {
 		log.Printf("[DEBUG] Skipping deployment of workers as services as swarmConfig is not set to run or swarm. Value: %#v", swarmConfig)
 		return
 	}
-
-
 	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 := getLocalIP()
+
+	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
@@ -594,7 +606,7 @@ func deployServiceWorkers(image string) {
 				Condition: swarm.RestartPolicyConditionOnFailure,
 			},
 			Placement: &swarm.Placement{
-				MaxReplicas: replicas,
+				Constraints: []string{},
 			},
 		},
 	}
@@ -1926,7 +1938,7 @@ func main() {
 
 		if len(containerId) > 0 {
 			pipelineUrl = "http://tenzir-node:5160"
-		} 
+		}
 
 		log.Printf("[WARNING] SHUFFLE_PIPELINE_URL not set, falling back to default URL: %s. If BASE_URL is set, we use the external IP for that", pipelineUrl)
 		os.Setenv("SHUFFLE_PIPELINE_URL", pipelineUrl)
@@ -2108,6 +2120,7 @@ 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)
 			hasStarted = true
 		}
 
@@ -2172,7 +2185,7 @@ func main() {
 
 						log.Printf("[ERROR] Failed handling pipeline (%s %s): %s. Deleting job anyway.", incRequest.Type, incRequest.ExecutionSource, err)
 					}
-			
+
 					toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
 				} else if incRequest.Type == "DOCKER_IMAGE_DOWNLOAD" {
 					log.Printf("[INFO] Should delete -> download new images: %#v", incRequest.ExecutionArgument)
@@ -2507,37 +2520,38 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed to create pipeline: %s", err)
 			return err
 		}
-	} else if incRequest.Type == "PIPELINE_DELETE" || incRequest.Type == "PIPELINE_STOP" { {
-		log.Printf("[INFO] Should delete pipeline %#v", identifier)
-		pipelineId, err := searchPipeline(identifier)
-		if err != nil {
-			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
-			return err
+	} else if incRequest.Type == "PIPELINE_DELETE" || incRequest.Type == "PIPELINE_STOP" {
+		{
+			log.Printf("[INFO] Should delete pipeline %#v", identifier)
+			pipelineId, err := searchPipeline(identifier)
+			if err != nil {
+				log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
+				return err
+			}
+
+			err = deletePipeline(pipelineId)
+			if err != nil {
+				log.Printf("[ERROR] Failed Deleting Pipeline %s", err)
+				return err
+			}
 		}
 
-		err = deletePipeline(pipelineId)
-		if err != nil {
-			log.Printf("[ERROR] Failed Deleting Pipeline %s", err)
-			return err
-		}
-	}
-		
-	/*
-	} else if incRequest.Type == "PIPELINE_STOP" {
-		log.Printf("[INFO] Should stop the pipeline %#v", identifier)
-		pipelineId, err := searchPipeline(identifier)
-		if err != nil {
-			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
-			toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
-			return err
-		}
-		_, err = updatePipelineState(command, pipelineId, "stop")
-		if err != nil {
-			log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err)
-			return err
-		} else {
-			log.Printf("[INFO] Successfully stopped the Pipeline: %s", pipelineId)
-		}
+		/*
+			} else if incRequest.Type == "PIPELINE_STOP" {
+				log.Printf("[INFO] Should stop the pipeline %#v", identifier)
+				pipelineId, err := searchPipeline(identifier)
+				if err != nil {
+					log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
+					toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
+					return err
+				}
+				_, err = updatePipelineState(command, pipelineId, "stop")
+				if err != nil {
+					log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err)
+					return err
+				} else {
+					log.Printf("[INFO] Successfully stopped the Pipeline: %s", pipelineId)
+				}
 		*/
 
 	} else if incRequest.Type == "PIPELINE_START" {
@@ -2678,17 +2692,17 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 
 	// Ensure restart policy is there
 	config := &container.Config{
-		Hostname:     containerName,
-		Cmd:          []string{"--commands=web server --mode=dev --bind=0.0.0.0"},
-		Image:        imageName,
-		Healthcheck:  healthconfig,
+		Hostname:    containerName,
+		Cmd:         []string{"--commands=web server --mode=dev --bind=0.0.0.0"},
+		Image:       imageName,
+		Healthcheck: healthconfig,
 		ExposedPorts: nat.PortSet{
 			"5160/tcp": struct{}{},
-			"514/udp": struct{}{},
-			"514/tcp": struct{}{},
+			"514/udp":  struct{}{},
+			"514/tcp":  struct{}{},
 		},
-		Entrypoint:   []string{containerName},
-		Env:		  []string{},
+		Entrypoint: []string{containerName},
+		Env:        []string{},
 	}
 
 	tenzirApikey := os.Getenv("TENZIR_PLUGINS__PLATFORM__API_KEY")
@@ -2698,39 +2712,37 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 	anyFound := false
 	if len(tenzirApikey) > 0 {
 		config.Env = append(config.Env, fmt.Sprintf("TENZIR_PLUGINS__PLATFORM__API_KEY=%s", tenzirApikey))
-		anyFound = true 
+		anyFound = true
 	}
 
 	if len(tenzirControlEndpoint) > 0 {
 		config.Env = append(config.Env, fmt.Sprintf("TENZIR_PLUGINS__PLATFORM__CONTROL_ENDPOINT=%s", tenzirControlEndpoint))
-		anyFound = true 
+		anyFound = true
 	}
 
 	if len(tenzirPluginsPlatform) > 0 {
 		config.Env = append(config.Env, fmt.Sprintf("TENZIR_PLUGINS__PLATFORM__TENANT_ID=%s", tenzirPluginsPlatform))
-		anyFound = true 
+		anyFound = true
 	}
 
 	tenzirStorageFolder := os.Getenv("SHUFFLE_STORAGE_FOLDER")
 	if len(tenzirStorageFolder) > 0 {
-		tenzirStorageFolder = tenzirStorageFolder 
+		tenzirStorageFolder = tenzirStorageFolder
 
 		if !strings.HasSuffix(tenzirStorageFolder, "/") {
 			tenzirStorageFolder = tenzirStorageFolder + "/"
 		}
 	} else {
 		tenzirStorageFolder = "/tmp/"
-		log.Printf("[DEBUG] Using folder %s for Tenzir storage. Change it using SHUFFLE_STORAGE_FOLDER", tenzirStorageFolder) 
+		log.Printf("[DEBUG] Using folder %s for Tenzir storage. Change it using SHUFFLE_STORAGE_FOLDER", tenzirStorageFolder)
 	}
 
-
 	if !anyFound {
-		//log.Printf("[DEBUG] No Tenzir Plugin environment variables found.") 
+		//log.Printf("[DEBUG] No Tenzir Plugin environment variables found.")
 	} else {
 		//log.Printf("[DEBUG] Attempting Tenzir connection with app.tenzir.com tenant '%s'", tenzirPluginsPlatform)
 	}
 
-
 	hostConfig := &container.HostConfig{
 		PortBindings: nat.PortMap{
 			"514/tcp":  []nat.PortBinding{{HostPort: "514"}},
@@ -3396,7 +3408,7 @@ func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
 		return pipelinePayload, nil
 	}
 
-	err := deployTenzirNode() 
+	err := deployTenzirNode()
 	if err != nil {
 		if (!strings.Contains(err.Error(), "SHUFFLE_SKIP_PIPELINES") && !strings.Contains(err.Error(), "Kubernetes not implemented for Tenzir node")) && !strings.Contains(err.Error(), "Tenzir Node is already running") && !strings.Contains(err.Error(), "docker daemon") {
 
@@ -3785,6 +3797,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
 		log.Printf("[ERROR] Failed reading body in worker request body to worker on %s: %s", streamUrl, err)
 		return err
 	}
+	window.AddEvent(time.Now())
 
 	if newresp.StatusCode != 200 {
 		log.Printf("[WARNING] POTENTIAL error running worker request (2) - status code is %d for %s, not 200. Body: %s", newresp.StatusCode, streamUrl, string(body))
@@ -3807,3 +3820,259 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
 	log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s. DEBUGGING:\n%s", workflowExecution.ExecutionId, streamUrl, debugCommand)
 	return nil
 }
+
+func AutoScale(ctx context.Context) {
+	if os.Getenv("SHUFFLE_SCALE_REPLICAS") != "" {
+		return
+	}
+
+	ticker := time.NewTicker(1 * time.Second)
+	coolDownPeriod := 10 * time.Second
+	queuePerMinuteInt = 20
+	if os.Getenv("SHUFFLE_QUEUE_PER_MINUTE") != "" {
+		var err error
+		queuePerMinuteInt, err = strconv.Atoi(os.Getenv("SHUFFLE_QUEUE_PER_MINUTE"))
+		if err != nil {
+			log.Printf("[WARNING] Cannot convert %s to int. Using default value for it: %d", queuePerMinute, queuePerMinuteInt)
+		}
+	}
+
+	lastScaleTime := time.Now()
+	currentWorkers := currentWokerCount(ctx, dockercli)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if time.Since(lastScaleTime) < (coolDownPeriod) {
+				continue
+			}
+			currentRequestCount := window.CountEvents(time.Now())
+			requiredReplicas := 0
+			if currentRequestCount >= queuePerMinuteInt*currentWorkers {
+				// FIXME: Hardcoded Max Replicas should be 6
+				requiredReplicas = int(math.Min(float64(6), float64(currentRequestCount/queuePerMinuteInt)+1))
+			}
+
+			if requiredReplicas > 0 {
+				err := scaleService(ctx, dockercli, uint64(requiredReplicas))
+				if err != nil {
+					log.Printf("[ERROR] Failed to scale service: %s", err)
+				} else {
+					lastScaleTime = time.Now()
+					currentWorkers = currentWokerCount(ctx, dockercli)
+				}
+			}
+		}
+	}
+}
+
+func scaleService(ctx context.Context, client *dockerclient.Client, replicas uint64) error {
+	service, _, err := client.ServiceInspectWithRaw(ctx, "shuffle-workers", types.ServiceInspectOptions{})
+	if err != nil {
+		return err
+	}
+
+	if service.Spec.Mode.Replicated == nil {
+		return errors.New("Service cannot be replicated")
+	}
+
+	if *service.Spec.Mode.Replicated.Replicas >= replicas {
+		return nil
+	}
+
+	service.Spec.Mode.Replicated.Replicas = &replicas
+
+	_, err = dockercli.ServiceUpdate(ctx, service.ID, service.Version, service.Spec, types.ServiceUpdateOptions{})
+	if err != nil {
+		return err
+	}
+
+	log.Printf("[INFO] Scaled shuffle-worker to %d replicas", replicas)
+	return nil
+}
+
+func currentWokerCount(ctx context.Context, client *dockerclient.Client) int {
+	service, _, err := client.ServiceInspectWithRaw(ctx, "shuffle-workers", types.ServiceInspectOptions{})
+	if err != nil {
+		return 0
+	}
+
+	if service.Spec.Mode.Replicated == nil {
+		return 0
+	}
+
+	return int(*service.Spec.Mode.Replicated.Replicas)
+}
+
+func queueScaleFactor(numQueue int, queuePerMin int) float64 {
+	if numQueue > queuePerMin {
+		queuePressure := float64(numQueue) / float64(queuePerMin)
+		return 1.0 + math.Min(queuePressure-1.0, 1.0)
+	}
+
+	return 1.0
+}
+
+func checkMemcached(ctx context.Context, dockercli *dockerclient.Client) (bool, error) {
+	containerName := "shuffle-cache"
+	continer, err := dockercli.ContainerInspect(context.Background(), containerName)
+	if err != nil {
+		if dockerclient.IsErrNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	}
+
+	if continer.State.Running == false {
+		log.Printf("[INFO] Container %s exists but is not running. Attempting to start it.", containerName)
+		err = dockercli.ContainerStart(ctx, containerName, container.StartOptions{})
+		if err != nil {
+			log.Printf("[ERROR] Failed to start container %s: %v", containerName, err)
+			return false, err
+		}
+		log.Printf("[INFO] Successfully started container %s.", containerName)
+		return true, nil
+	}
+
+	return continer.State.Running, nil
+}
+
+func deployMemcached(dockercli *dockerclient.Client) error {
+	if os.Getenv("SHUFFLE_MEMCACHED") != "" {
+		return errors.New("Memcached already running")
+	}
+
+	defaultMem := "1024"
+	log.Printf("[INFO] Spanning a default memcached container to handle the distribution between cache across different workers. Default memory assigned %s", defaultMem)
+
+	ctx := context.Background()
+
+	containerConfig := &container.Config{
+		Image: "memcached",
+		Cmd:   []string{"-m", defaultMem},
+	}
+
+	hostConfig := &container.HostConfig{
+		PortBindings: nat.PortMap{
+			"11211/tcp": []nat.PortBinding{{HostPort: "11211"}},
+		},
+	}
+
+	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
+	}
+	err = dockercli.ContainerStart(ctx, resp.ID, container.StartOptions{})
+	if err != nil {
+		log.Printf("[ERROR] Error starting memcached continer: %s", err)
+		return err
+	}
+
+	log.Printf("[INFO] Memcached container started successfully at port 11211")
+
+	return nil
+}
+
+// How do we get the cpu usage? maybe just get the number of requests (much more useful for apps)
+/*
+func nodesResourceUsage(ctx context.Context, client *dockerclient.Client) error {
+	nodes, err := client.NodeList(ctx, types.NodeListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, node := range nodes {
+		res := node.Description.Resources
+	}
+
+	return nil
+}
+*/
+
+/*
+func numberOfReplicas(ctx context.Context, queueLength int, config shuffle.ScalingConfig) (int, int) {
+	queueScaleFactor := queueScaleFactor(queueLength, config)
+	numReplicas := int(float64(queueLength) * queueScaleFactor)
+	serviceName := "shuffle-workers"
+	nodes, err := dockercli.NodeList(ctx, types.NodeListOptions{})
+	if err != nil {
+		log.Printf("[ERROR] Cannot find any nodes in the swarm network")
+	}
+
+	filterArgs := filters.NewArgs()
+	filterArgs.Add("service", serviceName)
+	filterArgs.Add("desired-state", "running")
+
+	tasks, err := dockercli.TaskList(context.Background(), types.TaskListOptions{
+		Filters: filterArgs,
+	})
+	if err != nil {
+		log.Fatalf("[WARNING] Failed to list tasks for service %s: %s", serviceName, err)
+	}
+
+	runningReplicas := len(tasks)
+	if numReplicas > runningReplicas*len(nodes) {
+		maxIncrease := config.MaxScaleUpStep
+		if numReplicas > (runningReplicas*len(nodes) + maxIncrease) {
+			numReplicas = runningReplicas + maxIncrease
+		}
+	}
+
+	if numReplicas < config.MinReplicas {
+		numReplicas = config.MinReplicas
+	}
+	if numReplicas > config.MaxReplicas {
+		numReplicas = config.MaxReplicas
+	}
+
+	return numReplicas, runningReplicas
+}
+*/
+
+// TODO: Currently we use number of request made for the worker to run a execution as it is much
+// easier to track in a window time frame. But this could be useful.
+func collectMetrics(ctx context.Context, dockerClient *dockerclient.Client) (int, error) {
+	client := shuffle.GetExternalClient(baseUrl)
+	fullUrl := fmt.Sprintf("%s/api/v1/workflows/queue", baseUrl)
+	req, err := http.NewRequest("GET", fullUrl, nil)
+	if err != nil {
+		log.Printf("[ERROR] Failed to send a request to %s: %s", fullUrl, err)
+		return 0, err
+	}
+
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("Org-Id", environment)
+	if len(auth) > 0 {
+		req.Header.Add("Authorization", auth)
+	}
+
+	if len(org) > 0 {
+		req.Header.Add("Org", org)
+	}
+
+	if len(orborusLabel) > 0 {
+		log.Printf("[DEBUG] Sending with Label '%s'", orborusLabel)
+		req.Header.Add("X-Orborus-Label", orborusLabel)
+	}
+
+	if swarmConfig != "run" && swarmConfig != "swarm" {
+		req.Header.Add("X-Orborus-Runmode", "Default")
+	} else {
+		req.Header.Add("X-Orborus-Runmode", "Docker Swarm")
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return 0, err
+	}
+
+	var executionRequests shuffle.ExecutionRequestWrapper
+	body, err := ioutil.ReadAll(resp.Body)
+
+	json.Unmarshal(body, &executionRequests)
+
+	return len(executionRequests.Data), nil
+}

functions/onprem/worker/worker.go

@@ -58,6 +58,7 @@ var dockerApiVersion = strings.ToLower(os.Getenv("DOCKER_API_VERSION"))
 
 var baseimagename = "frikky/shuffle"
 var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
+var executionCount int64
 
 // var baseimagename = os.Getenv("SHUFFLE_BASE_IMAGE_NAME")
 
@@ -70,6 +71,7 @@ var requestsSent = 0
 var appsInitialized = false
 
 var hostname string
+var maxReplicas = uint64(12)
 
 /*
 var environments []string
@@ -98,6 +100,8 @@ var finishedExecutions []string
 var imagesDistributed []string
 var imagedownloadTimeout = time.Second * 300
 
+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",
@@ -105,6 +109,7 @@ var autoDeploy = map[string]string{
 	"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.1.0": "frikky/shuffle:shuffle-subflow_1.1.0",
+	"shuffle-tools-fork:1.0.0": "frikky/shuffle:shuffle-tools-fork_1.0.0",
 }
 
 //"testing:1.0.0":         "frikky/shuffle:testing_1.0.0",
@@ -425,7 +430,7 @@ func deployk8sApp(image string, identifier string, env []string) error {
 
 	baseDeployMode := false
 
-	// check if autoDeploy contains a value 
+	// check if autoDeploy contains a value
 	// that is equal to the image being deployed.
 	for _, value := range autoDeploy {
 		if value == image {
@@ -2313,6 +2318,10 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 	// 4. Push to db
 	// IF FAIL: Set executionstatus: abort or cancel
 	ctx := context.Background()
+	if actionResult.ExecutionId == "TBD" {
+		return
+	}
+
 	workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId)
 	if err != nil {
 		log.Printf("[ERROR][%s] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, actionResult.ExecutionId, err)
@@ -3048,8 +3057,7 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl
 				Condition: swarm.RestartPolicyConditionAny,
 			},
 			Placement: &swarm.Placement{
-				// Max per node
-				MaxReplicas: replicatedJobs,
+				Constraints: []string{},
 			},
 		},
 	}
@@ -3503,6 +3511,10 @@ func baseDeploy() {
 			fmt.Sprintf("SHUFFLE_LOGS_DISABLED=%s", logsDisabled),
 		}
 
+		if key == "shuffle-tools-fork:1.0.0" {
+			env = append(env, fmt.Sprintf("SHUFFLE_ALLOW_PACKAGE_INSTALL=%s", "true"))
+		}
+
 		if strings.ToLower(os.Getenv("SHUFFLE_PASS_APP_PROXY")) == "true" {
 			//log.Printf("APPENDING PROXY TO THE APP!")
 			env = append(env, fmt.Sprintf("HTTP_PROXY=%s", os.Getenv("HTTP_PROXY")))
@@ -3844,6 +3856,8 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) {
 		time.Sleep(time.Duration(30) * time.Second)
 		checkUnfinished(resp, request, execRequest)
 	}()
+	window.AddEvent(time.Now())
+
 	ctx := context.Background()
 
 	// FIXME: This should be PER EXECUTION
@@ -4087,6 +4101,34 @@ func runWebserver(listener net.Listener) {
 		log.Printf("[DEBUG] Running webserver config for SWARM and K8s")
 	}
 	/*** ENDREMOVE ***/
+	var dockercli *dockerclient.Client
+	ctx := context.Background()
+	scaleReplicas := os.Getenv("SHUFFLE_APP_REPLICAS")
+	if len(scaleReplicas) > 0 {
+		tmpInt, err := strconv.Atoi(scaleReplicas)
+		if err != nil {
+			log.Printf("[ERROR] %s is not a valid number for replication", scaleReplicas)
+		} else {
+			maxReplicas = uint64(tmpInt)
+			_ = tmpInt
+		}
+
+		log.Printf("[DEBUG] SHUFFLE_APP_REPLICAS set to value %#v. Trying to overwrite default (%d/node)", scaleReplicas, maxReplicas)
+	}
+
+	maxExecutionsPerMinute := 10
+	if os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE") != "" {
+		tmpInt, err := strconv.Atoi(os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE"))
+		if err != nil {
+			log.Printf("[ERROR] %s is not a valid number for executions per minute", os.Getenv("SHUFFLE_APP_EXECUTIONS_PER_MINUTE"))
+		} else {
+			maxExecutionsPerMinute = tmpInt
+		}
+
+		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_DEBUG_MEMORY")) == "true" {
 		r.HandleFunc("/debug/pprof/", pprof.Index)
@@ -4121,3 +4163,205 @@ func runWebserver(listener net.Listener) {
 		log.Printf("[ERROR] Serve issue in worker: %#v", err)
 	}
 }
+
+func AutoScaleApps(ctx context.Context, client *dockerclient.Client, maxExecutionsPerMinute int) {
+	ticker := time.NewTicker(1 * time.Second)
+
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-ticker.C:
+			count := window.CountEvents(time.Now())
+			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))
+			}
+		}
+	}
+}
+
+func scaleApps(ctx context.Context, client *dockerclient.Client, replicas uint64) error {
+	client, err := dockerclient.NewEnvClient()
+	services, err := client.ServiceList(ctx, types.ServiceListOptions{})
+	if err != nil {
+		log.Printf("[ERROR] Failed to find services in the swarm: %s", err)
+	}
+
+	networkId, err := getNetworkId(ctx, client)
+	if err != nil {
+		log.Printf("[ERROR] Failed to get network Id in the swarm service: %s", err)
+	}
+
+	workers := numberOfWorkers(ctx, client)
+	if replicas > uint64(workers) {
+		return nil
+	}
+
+	for _, service := range services {
+		if service.Spec.Name == "shuffle-workers" {
+			continue
+		}
+
+		inNetwork := false
+		for _, vip := range service.Endpoint.VirtualIPs {
+			if vip.NetworkID == networkId {
+				inNetwork = true
+				break
+			}
+		}
+		if !inNetwork {
+			continue // skip services not in the target network
+		}
+
+		if service.Spec.Mode.Replicated == nil {
+			return errors.New("Service is not replicated")
+		}
+
+		if *service.Spec.Mode.Replicated.Replicas >= replicas {
+			continue
+		}
+
+		service.Spec.Mode.Replicated.Replicas = &replicas
+		_, err = client.ServiceUpdate(ctx, service.ID, service.Version, service.Spec, types.ServiceUpdateOptions{})
+		if err != nil {
+			return err
+		}
+
+	}
+
+	log.Printf("[DEBUG] Scaled all services to %d replicas", replicas)
+	return nil
+}
+
+func getNetworkId(ctx context.Context, dockercli *dockerclient.Client) (string, error) {
+	networkFilter := filters.NewArgs()
+	networkFilter.Add("name", swarmNetworkName)
+
+	networks, err := dockercli.NetworkList(ctx, types.NetworkListOptions{
+		Filters: networkFilter,
+	})
+
+	if err != nil || len(networks) == 0 {
+		return "", err
+	}
+	networkId := networks[0].ID
+
+	return networkId, nil
+}
+
+func numberOfApps(ctx context.Context, dockercli *dockerclient.Client) int {
+	// swarmNetworkName
+
+	var err error
+	if swarmNetworkName == "" {
+		swarmNetworkName = "shuffle_swarm_executions"
+	}
+
+	if dockercli == nil {
+		dockercli, err = dockerclient.NewEnvClient()
+		if err != nil {
+			log.Printf("[ERROR] Unable to create docker client (5): %s", err)
+			return 0
+		}
+	}
+
+	networkFilter := filters.NewArgs()
+	networkFilter.Add("name", swarmNetworkName)
+
+	networks, err := dockercli.NetworkList(ctx, types.NetworkListOptions{
+		Filters: networkFilter,
+	})
+
+	if err != nil || len(networks) == 0 {
+		return 0
+	}
+
+	networkId, err := getNetworkId(ctx, dockercli)
+	if err != nil {
+		log.Printf("[WARNING] Failed to get networkID is worker running in swarm: %s", err)
+		return 0
+	}
+
+	services, err := dockercli.ServiceList(ctx, types.ServiceListOptions{})
+	if err != nil {
+		log.Printf("[WARNING] Can't found any services. %s", err)
+		return 0
+	}
+
+	runningReplicas := 0
+
+	for _, service := range services {
+		if service.Spec.Name == "shuffle-workers" {
+			continue
+		}
+
+		inNetwork := false
+		for _, vip := range service.Endpoint.VirtualIPs {
+			if vip.NetworkID == networkId {
+				inNetwork = true
+				break
+			}
+		}
+		if !inNetwork {
+			continue // skip services not in the target network
+		}
+
+		filterArgs := filters.NewArgs()
+		filterArgs.Add("service", service.Spec.Name)
+		filterArgs.Add("desired-state", "running")
+
+		task, err := dockercli.TaskList(ctx, types.TaskListOptions{
+			Filters: filterArgs,
+		})
+		if err != nil {
+			log.Printf("[WARNING] Failed to get the list of running services %s: %s", service.Spec.Name, err)
+			continue
+		}
+
+		runningReplicas = len(task)
+		break
+	}
+
+	return runningReplicas
+}
+
+func IsServiceRunning(ctx context.Context, cli *dockerclient.Client) bool {
+	serviceName := "shuffle-tools_1-2-0"
+	filterArgs := filters.NewArgs()
+	filterArgs.Add("name", serviceName)
+
+	services, err := cli.ServiceList(ctx, types.ServiceListOptions{Filters: filterArgs})
+
+	if err != nil {
+		log.Printf("[ERROR] Couldn't find %s service running got error: %s", serviceName, err)
+		return false
+	}
+	if len(services) > 0 {
+		return true
+	}
+
+	return false
+}
+
+func numberOfWorkers(ctx context.Context, cli *dockerclient.Client) int {
+	cli, err := dockerclient.NewEnvClient()
+	service, _, err := cli.ServiceInspectWithRaw(ctx, "shuffle-workers", types.ServiceInspectOptions{})
+	if err != nil {
+		return 0
+	}
+
+	if service.Spec.Mode.Replicated == nil {
+		return 0
+	}
+
+	replics := *service.Spec.Mode.Replicated.Replicas
+	return int(replics)
+}