From 356fbe6f35cdc69375b2399b93a08b57caeee6c9 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Mon, 5 Feb 2024 22:50:34 +0530 Subject: [PATCH] hotfix: fixing image errors --- functions/onprem/worker/worker.go | 104 ++++++++++++------------------ 1 file changed, 40 insertions(+), 64 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 4eb0dc7f..6b9edf24 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -3,7 +3,6 @@ package main import ( "github.com/shuffle/shuffle-shared" - "bytes" "context" "encoding/json" @@ -22,8 +21,8 @@ import ( "time" "github.com/docker/docker/api/types" - "github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/mount" dockerclient "github.com/docker/docker/client" // This is for automatic removal of certain code :) @@ -52,9 +51,7 @@ var cleanupEnv = strings.ToLower(os.Getenv("CLEANUP")) var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME") var dockerApiVersion = strings.ToLower(os.Getenv("DOCKER_API_VERSION")) -//var baseimagename = "frikky/shuffle" -var baseimagerepository = os.Getenv("SHUFFLE_BASE_IMAGE_REPOSITORY") -var baseimagename = fmt.Sprintf("%s/%s", baseimagerepository, os.Getenv("SHUFFLE_BASE_IMAGE_NAME")) +var baseimagename = "frikky/shuffle" // var baseimagename = "registry.hub.docker.com/frikky/shuffle" var registryName = "registry.hub.docker.com" @@ -80,6 +77,7 @@ var startAction string //var allLogs map[string]string //var containerIds []string var downloadedImages []string + type ImageDownloadBody struct { Image string `json:"image"` } @@ -90,7 +88,6 @@ type ImageRequest struct { var finishedExecutions []string - // 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", @@ -137,7 +134,6 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo return err } - handleExecutionResult(workflowExecution) validated := shuffle.ValidateFinished(ctx, -1, workflowExecution) if validated { @@ -177,7 +173,7 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo } } - if len(subflowId) == 0 { + if len(subflowId) == 0 { log.Printf("[DEBUG][%s] No waiting result found. Not polling", workflowExecution.ExecutionId) for _, action := range workflowExecution.Workflow.Actions { @@ -185,19 +181,17 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo workflowExecution.Workflow.Triggers = append(workflowExecution.Workflow.Triggers, shuffle.Trigger{ AppName: action.AppName, Parameters: action.Parameters, - ID: action.ID, + ID: action.ID, }) } } - for _, trigger := range workflowExecution.Workflow.Triggers { //log.Printf("[DEBUG] Found trigger %s", trigger.AppName) if trigger.AppName != "User Input" && trigger.AppName != "Shuffle Workflow" && trigger.AppName != "shuffle-subflow" { continue } - // check if it has wait for results in params wait := false for _, param := range trigger.Parameters { @@ -216,9 +210,9 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo //log.Printf("[DEBUG][%s] Found result %s", workflowExecution.ExecutionId, result.Action.ID) if result.Action.ID == trigger.ID && result.Status != "SUCCESS" && result.Status != "FAILURE" { //log.Printf("[DEBUG][%s] Found subflow result that is not handled. Waiting for results", workflowExecution.ExecutionId) - + subflowId = result.Action.ID - found = true + found = true break } } @@ -237,21 +231,20 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo if len(subflowId) > 0 { // Under rerun period timeout - timeComparison := 120 + timeComparison := 120 log.Printf("[DEBUG][%s] Starting polling for %d seconds to see if new subflow updates are found on the backend that are not handled. Subflow ID: %s", workflowExecution.ExecutionId, timeComparison, subflowId) timestart := time.Now() streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl) for { - err = handleSubflowPoller(ctx, workflowExecution, streamResultUrl, subflowId) + err = handleSubflowPoller(ctx, workflowExecution, streamResultUrl, subflowId) if err == nil { log.Printf("[DEBUG] Subflow is finished and we are breaking the thingy") - + if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" && workflowExecution.ExecutionSource != "default" { log.Printf("[DEBUG] Force shutdown of worker due to optimized run with webserver. Expecting reruns to take care of this") os.Exit(0) } - break } @@ -273,7 +266,6 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo return nil } - // removes every container except itself (worker) func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) { log.Printf("[DEBUG][%s] Shutdown (%s) started with reason %#v. Result amount: %d. ResultsSent: %d, Send result: %#v, Parent: %#v", workflowExecution.ExecutionId, workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend, workflowExecution.ExecutionParent) @@ -313,7 +305,7 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason } */ } else { - + } if len(reason) > 0 && len(nodeId) > 0 { @@ -443,9 +435,9 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] RestartPolicy: "Never", Containers: []corev1.Container{ { - Name: value, - Image: image, - Env: buildEnvVars(envMap), + Name: value, + Image: image, + Env: buildEnvVars(envMap), // ImagePullPolicy: corev1.PullAlways, }, }, @@ -479,7 +471,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] if !strings.Contains(param.Value, "shuffle-backend") { continue - } + } // Automatic replacement as this is default if len(os.Getenv("BASE_URL")) > 0 { @@ -494,7 +486,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] } } - // Max 10% CPU every second //CPUShares: 128, //CPUQuota: 10000, @@ -521,7 +512,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] // Get environment for certificates volumeBinds := []string{} - volumeBindString:= os.Getenv("SHUFFLE_VOLUME_BINDS") + volumeBindString := os.Getenv("SHUFFLE_VOLUME_BINDS") if len(volumeBindString) > 0 { volumeBindSplit := strings.Split(volumeBindString, ",") for _, volumeBind := range volumeBindSplit { @@ -562,7 +553,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] Env: env, } - // Checking as late as possible, just in case. newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID) _, err := shuffle.GetCache(ctx, newExecId) @@ -842,7 +832,7 @@ func askOtherWorkersToDownloadImage(image string) { // Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" { log.Printf("[DEBUG] SHUFFLE_AUTO_IMAGE_DOWNLOAD is false. NOT distributing images %s", image) - return + return } urls, err := getWorkerURLs() @@ -866,7 +856,7 @@ func askOtherWorkersToDownloadImage(image string) { req, err := http.NewRequest( "POST", url, - bytes.NewBuffer(imageJSON), + bytes.NewBuffer(imageJSON), ) if err != nil { @@ -901,7 +891,6 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { return } - startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId) dockercli, err := dockerclient.NewEnvClient() @@ -968,7 +957,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { // marshal action and put it in there rofl //log.Printf("[INFO][%s] Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", workflowExecution.ExecutionId, action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters)) - + log.Printf("[DEBUG][%s] Action: Send, Label: '%s', Action: '%s', Run status: %s, Extra=", workflowExecution.ExecutionId, action.Label, action.AppName, workflowExecution.Status) actionData, err := json.Marshal(action) @@ -1054,10 +1043,9 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) > 0 { - env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))) + env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))) } - // Fixes issue: // standard_go init_linux.go:185: exec user process caused "argument list too long" // https://devblogs.microsoft.com/oldnewthing/20100203-00/?p=15083 @@ -1085,8 +1073,6 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion), } - - // If cleanup is set, it should run for efficiency pullOptions := types.ImagePullOptions{} if cleanupEnv == "true" { @@ -1585,7 +1571,7 @@ func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.Workflow } log.Printf("[INFO][%s] Status: %s, Results: %d, actions: %d", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra) - return errors.New("Subflow status not found yet") + return errors.New("Subflow status not found yet") } func handleDefaultExecutionWrapper(ctx context.Context, workflowExecution shuffle.WorkflowExecution, streamResultUrl string, extra int) error { @@ -1980,10 +1966,9 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Execution is not executing, but %s"}`, workflowExecution.Status))) } - log.Printf("[DEBUG][%s] Shutting down (35)", workflowExecution.ExecutionId) - // Force sending result + // Force sending result shutdownData, err := json.Marshal(workflowExecution) if err != nil { log.Printf("[ERROR][%s] Failed marshalling execution (35): %s", workflowExecution.ExecutionId, err) @@ -2058,11 +2043,11 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl attempts += 1 log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(parsedValue.Results), resultLength) /* - if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) { - log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) - runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) - return - } + if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) { + log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) + runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) + return + } */ } } @@ -2096,7 +2081,6 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl func sendSelfRequest(actionResult shuffle.ActionResult) { - data, err := json.Marshal(actionResult) if err != nil { log.Printf("[ERROR][%s] Shutting down (24): Failed to unmarshal data for backend: %s", actionResult.ExecutionId, err) @@ -2155,20 +2139,20 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { // Basically to reduce backend strain /* - if shuffle.ArrayContains(finishedExecutions, workflowExecution.ExecutionId) { - log.Printf("[INFO][%s] NOT sending backend info since it's already been sent before.", workflowExecution.ExecutionId) - return - } + if shuffle.ArrayContains(finishedExecutions, workflowExecution.ExecutionId) { + log.Printf("[INFO][%s] NOT sending backend info since it's already been sent before.", workflowExecution.ExecutionId) + return + } */ // Take it down again /* - if len(finishedExecutions) > 100 { - log.Printf("[DEBUG][%s] Removing old execution from finishedExecutions: %s", workflowExecution.ExecutionId, finishedExecutions[0]) - finishedExecutions = finishedExecutions[99:] - } + if len(finishedExecutions) > 100 { + log.Printf("[DEBUG][%s] Removing old execution from finishedExecutions: %s", workflowExecution.ExecutionId, finishedExecutions[0]) + finishedExecutions = finishedExecutions[99:] + } - finishedExecutions = append(finishedExecutions, workflowExecution.ExecutionId) + finishedExecutions = append(finishedExecutions, workflowExecution.ExecutionId) */ streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl) @@ -2222,7 +2206,7 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) bool { log.Printf("[INFO][%s] VALIDATION. Status: %s, shuffle.Actions: %d, Extra: %d, Results: %d. Parent: %#v", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results), workflowExecution.ExecutionParent) - if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1 && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions)+extra && len(workflowExecution.Workflow.Actions) > 0) { + if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1 && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions)+extra && len(workflowExecution.Workflow.Actions) > 0) { if workflowExecution.Status == "FINISHED" { for _, result := range workflowExecution.Results { @@ -2233,7 +2217,6 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) bool { } } - log.Printf("[DEBUG][%s] Should send full result to %s", workflowExecution.ExecutionId, baseUrl) //data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization) @@ -2317,7 +2300,6 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) { // GetLocalIP returns the non loopback local IP of the host func getLocalIP() string { - addrs, err := net.InterfaceAddrs() if err != nil { return "" @@ -2363,7 +2345,6 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener { log.Printf("[DEBUG] OLD HOSTNAME: %s", appCallbackUrl) - port := listener.Addr().(*net.TCPAddr).Port // Set the port environment variable os.Setenv("WORKER_PORT", fmt.Sprintf("%d", port)) @@ -2521,7 +2502,6 @@ func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) { */ } - // Runs data discovery func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution) error { @@ -2765,7 +2745,7 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe if newresp.StatusCode != 200 { log.Printf("[ERROR] %sStatusCode (1): %d", string(body), newresp.StatusCode) time.Sleep(time.Duration(sleepTime) * time.Second) - return environments, errors.New(fmt.Sprintf("Bad status code: %d", newresp.StatusCode) ) + return environments, errors.New(fmt.Sprintf("Bad status code: %d", newresp.StatusCode)) } err = json.Unmarshal(body, &workflowExecution) @@ -2848,7 +2828,6 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe // Set environment variable - //log.Printf("Before wait") //wg := sync.WaitGroup{} //wg.Add(1) @@ -2910,7 +2889,7 @@ func main() { timezone = "Europe/Amsterdam" } - if baseimagename == "" || baseimagename == "/" { + if baseimagename == "" { log.Printf("[DEBUG] Setting baseimagename") baseimagename = "frikky/shuffle" // Dockerhub //baseimagename = "shuffle" // Github (ghcr.io) @@ -2920,7 +2899,6 @@ func main() { swarmConfig := os.Getenv("SHUFFLE_SWARM_CONFIG") log.Printf("[INFO] Running with timezone %s and swarm config %#v", timezone, swarmConfig) - authorization := "" executionId := "" @@ -3225,7 +3203,6 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) { return } - for _, img := range images { for _, tag := range img.RepoTags { splitTag := strings.Split(tag, ":") @@ -3238,7 +3215,7 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) { possibleNames = append(possibleNames, fmt.Sprintf("frikky/shuffle:%s", baseTag)) possibleNames = append(possibleNames, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", baseTag)) - if (arrayContains(possibleNames, image.Image)) { + if arrayContains(possibleNames, image.Image) { log.Printf("[DEBUG] Image %s already downloaded that has been requested to download", image.Image) resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "image already present"}`))) @@ -3263,7 +3240,6 @@ func runWebserver(listener net.Listener) { r.HandleFunc("/api/v1/run", handleRunExecution).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/download", handleDownloadImage).Methods("POST", "OPTIONS") - if strings.ToLower(os.Getenv("SHUFFLE_DEBUG_MEMORY")) == "true" { r.HandleFunc("/debug/pprof/", pprof.Index) r.HandleFunc("/debug/pprof/heap", pprof.Handler("heap").ServeHTTP)