diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 34568564..6ba7ca41 100755 --- a/functions/onprem/orborus/orborus.go +++ b/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 +} diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index fa9895fa..ff0fb8ae 100644 --- a/functions/onprem/worker/worker.go +++ b/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", @@ -426,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 { @@ -2314,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) @@ -3049,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{}, }, }, } @@ -3849,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 @@ -4092,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) @@ -4126,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) +}