From 0e9b975c98b349a6a535fcb344211401060dbf53 Mon Sep 17 00:00:00 2001 From: Frikky Date: Wed, 3 Jan 2024 14:09:49 +0000 Subject: [PATCH] Updated worker nightly to run with latest shared version --- functions/onprem/worker/go.mod | 2 +- functions/onprem/worker/go.sum | 4 ++ functions/onprem/worker/worker.go | 81 ++++++++++++++++++++++--------- 3 files changed, 64 insertions(+), 23 deletions(-) diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index b847ac38..698af467 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -11,7 +11,7 @@ require ( github.com/gorilla/mux v1.8.0 github.com/patrickmn/go-cache v2.1.0+incompatible github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.5.31 + github.com/shuffle/shuffle-shared v0.5.53 k8s.io/api v0.28.3 k8s.io/apimachinery v0.28.3 k8s.io/client-go v0.28.3 diff --git a/functions/onprem/worker/go.sum b/functions/onprem/worker/go.sum index 821c92f0..64484995 100644 --- a/functions/onprem/worker/go.sum +++ b/functions/onprem/worker/go.sum @@ -288,6 +288,10 @@ github.com/shuffle/shuffle-shared v0.5.29 h1:n4vThl7v3mFVXbrIW71XREFdmZZo7mOBAWx github.com/shuffle/shuffle-shared v0.5.29/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI= github.com/shuffle/shuffle-shared v0.5.31 h1:OV4IIfKWWFW66WjGvyXOmmsSz3p8pW9L1ge1mDo8ftM= github.com/shuffle/shuffle-shared v0.5.31/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI= +github.com/shuffle/shuffle-shared v0.5.44 h1:6WiFPIsij+IWvXY7vzVX7cUicb+PYOzhTbWF/gDmYeU= +github.com/shuffle/shuffle-shared v0.5.44/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI= +github.com/shuffle/shuffle-shared v0.5.53 h1:Osr5sjr0wppqlf/MjH1oykdcw/Q48d8+ZnerRrYfOA4= +github.com/shuffle/shuffle-shared v0.5.53/go.mod h1:X613gbo0dT3fnYvXDRwjQZyLC+T49T2nSQOrCV5QMlI= github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e h1:MRM5ITcdelLK2j1vwZ3Je0FKVCfqOLp5zO6trqMLYs0= github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e/go.mod h1:XV66xRDqSt+GTGFMVlhk3ULuV0y9ZmzeVGR4mloJI3M= github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 0c8fd3a5..2eaa1f74 100755 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -354,7 +354,8 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason req.Header.Add("Content-Type", "application/json") //log.Printf("[DEBUG][%s] All App Logs: %#v", workflowExecution.ExecutionId, allLogs) - newresp, err := topClient.Do(req) + client := shuffle.GetExternalClient(abortUrl) + newresp, err := client.Do(req) if err != nil { log.Printf("[WARNING][%s] Failed abort request: %s", workflowExecution.ExecutionId, err) } else { @@ -464,6 +465,9 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] // Check action if subflow // Check if url is default (shuffle-backend) // If it doesn't exist, add it + + // FIXME: This does NOT replace it in all cases as the data + // is not saved in the database as the correct param. if action.AppName == "shuffle-subflow" { // Automatic replacement of URL for paramIndex, param := range action.Parameters { @@ -471,17 +475,19 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] continue } - if strings.Contains(param.Value, "shuffle-backend") { - // Automatic replacement as this is default - if len(os.Getenv("BASE_URL")) > 0 { - action.Parameters[paramIndex].Value = os.Getenv("BASE_URL") - log.Printf("[DEBUG][%s] Replaced backend_url with base_url %s", workflowExecution.ExecutionId, os.Getenv("BASE_URL")) - } + if !strings.Contains(param.Value, "shuffle-backend") { + continue + } - if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 { - action.Parameters[paramIndex].Value = os.Getenv("SHUFFLE_CLOUDRUN_URL") - log.Printf("[DEBUG][%s] Replaced backend_url with cloudrun %s", workflowExecution.ExecutionId, os.Getenv("SHUFFLE_CLOUDRUN_URL")) - } + // Automatic replacement as this is default + if len(os.Getenv("BASE_URL")) > 0 { + action.Parameters[paramIndex].Value = os.Getenv("BASE_URL") + log.Printf("[DEBUG][%s] Replaced backend_url with base_url %s", workflowExecution.ExecutionId, os.Getenv("BASE_URL")) + } + + if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 { + action.Parameters[paramIndex].Value = os.Getenv("SHUFFLE_CLOUDRUN_URL") + log.Printf("[DEBUG][%s] Replaced backend_url with cloudrun %s", workflowExecution.ExecutionId, os.Getenv("SHUFFLE_CLOUDRUN_URL")) } } } @@ -745,6 +751,8 @@ func removeContainer(containername string) error { return err } + defer cli.Close() + // FIXME - ucnomment // containers, err := cli.ContainerList(ctx, types.ContainerListOptions{ // All: true, @@ -800,6 +808,8 @@ func getWorkerURLs() ([]string, error) { return workerUrls, err } + defer cli.Close() + // Specify the name of the service for which you want to list tasks serviceName := "shuffle-workers" @@ -827,12 +837,19 @@ func askOtherWorkersToDownloadImage(image string) { return } + // 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 + } + urls, err := getWorkerURLs() if err != nil { log.Printf("[ERROR] Error in listing worker urls: %s", err) return } + httpClient := &http.Client{} for _, url := range urls { log.Printf("[DEBUG] Trying to speak to: %s", url) imagesRequest := ImageRequest{ @@ -855,7 +872,6 @@ func askOtherWorkersToDownloadImage(image string) { continue } - httpClient := &http.Client{} resp, err := httpClient.Do(req) if err != nil { log.Printf("[ERROR] Error in making request to %s : %s", url, err) @@ -892,6 +908,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { return } + defer dockercli.Close() + for _, action := range relevantActions { appname := action.AppName appversion := action.AppVersion @@ -1116,6 +1134,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true) return } else { + defer reader.Close() baseTag := strings.Split(image, ":") if len(baseTag) > 1 { tag := baseTag[1] @@ -1226,6 +1245,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true) return } else { + defer reader.Close() baseTag := strings.Split(image, ":") if len(baseTag) > 1 { tag := baseTag[1] @@ -1494,7 +1514,8 @@ func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.Workflow bytes.NewBuffer([]byte(data)), ) - newresp, err := topClient.Do(req) + client := shuffle.GetExternalClient(streamResultUrl) + newresp, err := client.Do(req) if err != nil { log.Printf("[ERROR] Failed making request (1): %s", err) time.Sleep(time.Duration(sleepTime) * time.Second) @@ -2101,7 +2122,8 @@ func sendSelfRequest(actionResult shuffle.ActionResult) { return } - newresp, err := topClient.Do(req) + client := shuffle.GetExternalClient(streamUrl) + newresp, err := client.Do(req) if err != nil { log.Printf("[ERROR][%s] Error running finishing request (2): %s", actionResult.ExecutionId, err) return @@ -2158,9 +2180,10 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { return } - newresp, err := topClient.Do(req) + client := shuffle.GetExternalClient(streamUrl) + newresp, err := client.Do(req) if err != nil { - log.Printf("[ERROR][%s] Error running finishing request: %s", workflowExecution.ExecutionId, err) + log.Printf("[ERROR][%s] Error running finishing request (1): %s", workflowExecution.ExecutionId, err) log.Printf("[DEBUG][%s] Shutting down (23)", workflowExecution.ExecutionId) shutdown(workflowExecution, "", "", false) return @@ -2348,6 +2371,12 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener { func downloadDockerImageBackend(client *http.Client, imageName string) error { log.Printf("[DEBUG] Trying to download image %s from backend %s as it doesn't exist. All images: %#v", imageName, baseUrl, downloadedImages) + // Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD + if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" { + log.Printf("[DEBUG] SHUFFLE_AUTO_IMAGE_DOWNLOAD is false. Not downloading image %s", imageName) + return nil + } + if arrayContains(downloadedImages, imageName) { log.Printf("[DEBUG] Image %s already downloaded", imageName) return nil @@ -2407,12 +2436,15 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error { return err } + defer dockercli.Close() + imageLoadResponse, err := dockercli.ImageLoad(context.Background(), tar, true) if err != nil { log.Printf("[ERROR] Error loading images: %s", err) return err } + defer imageLoadResponse.Body.Close() body, err := ioutil.ReadAll(imageLoadResponse.Body) if err != nil { log.Printf("[ERROR] Error reading: %s", err) @@ -2570,9 +2602,11 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, log.Printf("[DEBUG][%s] Adding %s to cache (%#v)", workflowExecution.ExecutionId, newExecId, action.Name) } - // FIXME: Add 5 tries + client := shuffle.GetExternalClient(streamUrl) - newresp, err := topClient.Do(req) + // Set client timeout to 5 seconds + client.Timeout = time.Duration(10) * time.Second + newresp, err := client.Do(req) if err != nil { // Another timeout issue here somewhere // context deadline @@ -2630,6 +2664,8 @@ func baseDeploy() { return } + defer cli.Close() + for key, value := range autoDeploy { newNameSplit := strings.Split(key, ":") @@ -2897,7 +2933,6 @@ func main() { shutdown(workflowExecution, "", "", true) } - topClient = client firstRequest := true environments := []string{} for { @@ -3007,15 +3042,14 @@ func handleRunExecution(resp http.ResponseWriter, request *http.Request) { var workflowExecution shuffle.WorkflowExecution data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, execRequest.ExecutionId, execRequest.Authorization) streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl) - topClient = shuffle.GetExternalClient(streamResultUrl) - req, err := http.NewRequest( "POST", streamResultUrl, bytes.NewBuffer([]byte(data)), ) - newresp, err := topClient.Do(req) + client := shuffle.GetExternalClient(streamResultUrl) + newresp, err := client.Do(req) if err != nil { log.Printf("[ERROR] Failed making request (2): %s", err) resp.WriteHeader(401) @@ -3145,6 +3179,8 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) { return } + defer client.Close() + // check if images are already downloaded // Retrieve a list of Docker images images, err := client.ImageList(context.Background(), types.ImageListOptions{}) @@ -3155,6 +3191,7 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) { return } + for _, img := range images { for _, tag := range img.RepoTags { splitTag := strings.Split(tag, ":")