From d74fab7bf88e5facf27b809e470d463abc0f9d1e Mon Sep 17 00:00:00 2001 From: frikky Date: Sun, 19 Dec 2021 02:37:34 +0100 Subject: [PATCH] Fixed more bugs related to microservice setup --- docker-compose.yml | 2 +- functions/onprem/worker/worker.go | 87 ++++++++++++++++++++++++++++++- 2 files changed, 86 insertions(+), 3 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 1bcdf88b..3808caa8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -59,7 +59,7 @@ services: - HTTPS_PROXY=${HTTPS_PROXY} - SHUFFLE_PASS_WORKER_PROXY=${SHUFFLE_PASS_WORKER_PROXY} - SHUFFLE_PASS_APP_PROXY=${SHUFFLE_PASS_APP_PROXY} - - SHUFFLE_SCALE_REPLICAS=5 + - SHUFFLE_SCALE_REPLICAS=1 - SHUFFLE_SWARM_CONFIG=runn - SHUFFLE_SWARM_NETWORK_NAME=shuffle-executions restart: unless-stopped diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index c2064cce..f75905a0 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -74,6 +74,15 @@ var containerIds []string var downloadedImages []string var executedIds = []string{} +var finishedIds = []string{} + +// Images to be autodeployed in the latest version of Shuffle. +var autoDeploy = map[string]string{ + "shuffle-subflow:1.0.0": "frikky/shuffle:shuffle-subflow_1.0.0", + "http:1.1.0": "frikky/shuffle:http_1.1.0", + "shuffle-tools:1.1.0": "frikky/shuffle:shuffle-tools_1.1.0", + "testing:1.0.0": "frikky/shuffle:testing_1.0.0", +} //if !shuffle.ArrayContains(executedIds, //fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID) @@ -739,7 +748,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { // care if it gets stuck in a loop. // FIXME: Force killing a worker should result in a notification somewhere if len(nextActions) == 0 { - log.Printf("[INFO] No next action. Finished? Result vs shuffle.Actions: %d - %d", len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) + log.Printf("[INFO][%s] No next action. Finished? Result vs shuffle.Actions: %d - %d", workflowExecution.ExecutionId, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) exit := true for _, item := range workflowExecution.Results { if item.Status == "EXECUTING" { @@ -749,7 +758,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } if len(environments) == 1 { - log.Printf("[INFO] Should send results to the backend because environments are %s", environments) + log.Printf("[INFO][%s] Should send results to the backend because environments are %s", workflowExecution.ExecutionId, environments) validateFinished(workflowExecution) } @@ -2220,6 +2229,10 @@ func getLocalIP() string { log.Printf("[DEBUG] Found hostname %s since worker is running with \"run\" command", name) return name + /** + Everything below was a test to see if we needed to match directly to a network interface. May require docker network API. + **/ + log.Printf("[DEBUG] Looking for IP for the external docker-network %s", swarmNetworkName) // Different process to ensure we find the right IP. // Necessary due to Ingress being added to docker ser @@ -2769,6 +2782,64 @@ func sendAppRequest(incomingUrl, appName string, port int, action shuffle.Action return nil } +// Function to auto-deploy certain apps if "run" is set +// Has some issues with loading when running multiple workers and such. +func baseDeploy() { + return + + cli, err := dockerclient.NewEnvClient() + if err != nil { + log.Printf("[ERROR] Unable to create docker client (3): %s", err) + return + } + + for key, value := range autoDeploy { + newNameSplit := strings.Split(key, ":") + + action := shuffle.Action{ + AppName: newNameSplit[0], + AppVersion: newNameSplit[1], + ID: "TBD", + } + + workflowExecution := shuffle.WorkflowExecution{ + ExecutionId: "TBD", + } + + appname := action.AppName + appversion := action.AppVersion + appname = strings.Replace(appname, ".", "-", -1) + appversion = strings.Replace(appversion, ".", "-", -1) + + env := []string{ + fmt.Sprintf("EXECUTIONID=%s", workflowExecution.ExecutionId), + fmt.Sprintf("AUTHORIZATION=%s", workflowExecution.Authorization), + fmt.Sprintf("CALLBACK_URL=%s", baseUrl), + fmt.Sprintf("BASE_URL=%s", appCallbackUrl), + fmt.Sprintf("TZ=%s", timezone), + } + + 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"))) + env = append(env, fmt.Sprintf("HTTPS_PROXY=%s", os.Getenv("HTTPS_PROXY"))) + } + + identifier := fmt.Sprintf("%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId) + if strings.Contains(identifier, " ") { + identifier = strings.ReplaceAll(identifier, " ", "-") + } + + //deployApp(cli, value, identifier, env, workflowExecution, action) + log.Printf("[DEBUG] Deploying app with identifier %s to ensure basic apps are available from the get-go", identifier) + go deployApp(cli, value, identifier, env, workflowExecution, action) + //err := deployApp(cli, value, identifier, env, workflowExecution, action) + //if err != nil { + // log.Printf("[DEBUG] Failed deploying app %s: %s", value, err) + //} + } +} + // Initial loop etc func main() { /* @@ -2825,6 +2896,11 @@ func main() { // Forcing download just in case on the first iteration. workflowExecution := shuffle.WorkflowExecution{} + //var autoDeploy = []string{"frikky/shuffle:shuffle-subflow_1.0.0", "frikky/shuffle:http_1.1.0", "frikky/shuffle:shuffle-tools_1.1.0", "frikky/shuffle:testing_1.0.0"} + + //go baseDeploy() + baseDeploy() + listener := webserverSetup(workflowExecution) runWebserver(listener) log.Printf("[ERROR] Stopped listener %#v - exiting.", listener) @@ -3013,7 +3089,14 @@ func main() { } } +func checkUnfinishedExecutions() { + // Meant as a function that periodically checks whether previous executions have finished or not. + // Should probably be based on executedIds and finishedIds +} + func handleRunExecution(resp http.ResponseWriter, request *http.Request) { + checkUnfinishedExecutions() + body, err := ioutil.ReadAll(request.Body) if err != nil { log.Println("[WARNING] Failed reading body for stream result queue")