Made docker image download controllable in shuffle-shared vs just worker

Author: frikky

codegen.go

@@ -16,14 +16,17 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"net/http"
 
 	"cloud.google.com/go/storage"
 	"github.com/frikky/kin-openapi/openapi3"
+	docker "github.com/docker/docker/client"
 
 	//"github.com/satori/go.uuid"
 	"gopkg.in/yaml.v2"
 )
 
+var downloadedImages = []string{}
 var pythonAllowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_"
 var pythonReplacements = map[string]string{
 	"[": "",
@@ -3771,3 +3774,125 @@ func RemoveJsonValues(input []byte, depth int64) ([]byte, string, error) {
 
 	return input, keyToken, nil
 }
+
+func DownloadDockerImageBackend(topClient *http.Client, imageName string) error {
+	// 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) && project.Environment == "worker" {
+		log.Printf("[DEBUG] Image %s already downloaded - not re-downloading. This only applies to workers.", imageName)
+		return nil
+	}
+
+	baseUrl := os.Getenv("BASE_URL")
+	log.Printf("[DEBUG] Trying to download image %s from backend %s as it doesn't exist. All images: %#v", imageName, baseUrl, downloadedImages)
+
+	downloadedImages = append(downloadedImages, imageName)
+
+	data := fmt.Sprintf(`{"name": "%s"}`, imageName)
+	dockerImgUrl := fmt.Sprintf("%s/api/v1/get_docker_image", baseUrl)
+
+	req, err := http.NewRequest(
+		"POST",
+		dockerImgUrl,
+		bytes.NewBuffer([]byte(data)),
+	)
+
+	// Specific to the worker
+	authorization := os.Getenv("AUTHORIZATION")
+	if len(authorization) > 0 {
+		req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization))
+	} else {
+		// Specific to Orborus auth (org + auth) -> environment auth
+		authorization = os.Getenv("AUTH")
+		if len(authorization) > 0 {
+			log.Printf("[DEBUG] Found Orborus environment auth - adding to header.")
+			req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization))
+
+			org := os.Getenv("ORG")
+			if len(org) > 0 {
+				req.Header.Add("Org-Id", org)
+			}
+
+		} else {
+			log.Printf("[WARNING] No auth found - running backend download without it.")
+		}
+	}
+
+	newresp, err := topClient.Do(req)
+	if err != nil {
+		log.Printf("[ERROR] Failed download request for %s: %s", imageName, err)
+		return err
+	}
+
+	defer newresp.Body.Close()
+	if newresp.StatusCode != 200 {
+		log.Printf("[ERROR] Docker download for image %s (backend) StatusCode (1): %d", imageName, newresp.StatusCode)
+		return errors.New(fmt.Sprintf("Failed to get image - status code %d", newresp.StatusCode))
+	}
+
+	newImageName := strings.Replace(imageName, "/", "_", -1)
+	newFileName := newImageName + ".tar"
+
+	tar, err := os.Create(newFileName)
+	if err != nil {
+		log.Printf("[WARNING] Failed creating file: %s", err)
+		return err
+	}
+
+	defer tar.Close()
+	_, err = io.Copy(tar, newresp.Body)
+	if err != nil {
+		log.Printf("[WARNING] Failed response body copying: %s", err)
+		return err
+	}
+	tar.Seek(0, 0)
+
+	dockercli, err := docker.NewEnvClient()
+	if err != nil {
+		log.Printf("[ERROR] Unable to create docker client (3): %s", err)
+		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)
+		return err
+	}
+
+	if strings.Contains(string(body), "no such file") {
+		return errors.New(string(body))
+	}
+
+	baseTag := strings.Split(imageName, ":")
+	if len(baseTag) > 1 {
+		tag := baseTag[1]
+		log.Printf("[DEBUG] Creating tag copies of downloaded containers from tag %s", tag)
+
+		// Remapping
+		ctx := context.Background()
+		dockercli.ImageTag(ctx, imageName, fmt.Sprintf("frikky/shuffle:%s", tag))
+		dockercli.ImageTag(ctx, imageName, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", tag))
+
+		downloadedImages = append(downloadedImages, fmt.Sprintf("frikky/shuffle:%s", tag))
+		downloadedImages = append(downloadedImages, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", tag))
+
+	}
+
+	os.Remove(newFileName)
+
+	log.Printf("[INFO] Successfully loaded image %s: %s", imageName, string(body))
+	return nil
+}

db-connector.go

@@ -3724,7 +3724,6 @@ func GetOrg(ctx context.Context, id string) (*Org, error) {
 	} else {
 		key := datastore.NameKey(nameKey, id, nil)
 		if err := project.Dbclient.Get(ctx, key, curOrg); err != nil {
-			log.Printf("[ERROR] Error in org loading (2) for %s: %s", key, err)
 			//log.Printf("Users: %s", curOrg.Users)
 			if strings.Contains(err.Error(), `cannot load field`) && strings.Contains(err.Error(), `users`) && !strings.Contains(err.Error(), `users_last_session`) {
 				//Self correcting Org handler for user migration. This may come in handy if we change the structure of private apps later too.
@@ -3756,9 +3755,10 @@ func GetOrg(ctx context.Context, id string) (*Org, error) {
 					setOrg = true
 				}
 			} else if strings.Contains(err.Error(), `cannot load field`) {
-				log.Printf("[WARNING] Error in org loading (4), but returning without warning: %s", err)
+				//log.Printf("[WARNING] Error in org loading (4), but returning without warning: %s", err)
 				err = nil
 			} else {
+				log.Printf("[ERROR] Error in org loading (2) for %s: %s", key, err)
 				return &Org{}, err
 			}
 		}

oauth2.go

@@ -3534,7 +3534,7 @@ func GetOauth2ApplicationPermissionToken(ctx context.Context, user User, appAuth
 	username := ""
 	password := ""
 
-	log.Printf("[DEBUG] Got %d auth fields (%s)", len(appAuth.Fields), appAuth.Id)
+	//log.Printf("[DEBUG] Got %d auth fields (%s)", len(appAuth.Fields), appAuth.Id)
 	for _, field := range appAuth.Fields {
 		if field.Key == "client_secret" {
 			clientSecret = field.Value
@@ -3580,22 +3580,15 @@ func GetOauth2ApplicationPermissionToken(ctx context.Context, user User, appAuth
 		refreshData += fmt.Sprintf("&scope=%s", strings.Replace(scope, ",", " ", -1))
 	}
 
-
 	if strings.Contains(refreshData, "user_impersonation") && strings.Contains (refreshData, "azure") && !strings.Contains(refreshData, "resource="){
 		// Add "resource" for microsoft hings
 		refreshData += "&resource=https://management.azure.com"
 	}
 
-	/*
-	if strings.Contains(refreshData, "client_credentials") && strings.Contains (refreshData, "azure") && !strings.Contains(refreshData, "resource="){
-		refreshData += "&resource=https://management.azure.com"
-	}
-	*/
-
+	// Not necessary for refresh
 	log.Printf("[DEBUG] Oauth2 REFRESH DATA: %#v. URL: %#v", refreshData, tokenUrl)
 
 	client := GetExternalClient(tokenUrl)
-
 	req, err := http.NewRequest(
 		"POST",
 		tokenUrl,
@@ -3619,14 +3612,15 @@ func GetOauth2ApplicationPermissionToken(ctx context.Context, user User, appAuth
 		return appAuth, err
 	}
 
-	log.Printf("[DEBUG] Oauth2 application auth Response for %s: %d", tokenUrl, newresp.StatusCode)
-
 	defer newresp.Body.Close()
 	body, err := ioutil.ReadAll(newresp.Body)
 	if err != nil {
+		log.Printf("[ERROR] Oauth2 application auth: Failed to read response body: %s", err)
 		return appAuth, err
 	}
 
+	log.Printf("[DEBUG] Oauth2 application auth Response for %s: %d", tokenUrl, newresp.StatusCode)
+
 	if newresp.StatusCode >= 300 {
 		// Printing on error to handle in future instances
 		log.Printf("[ERROR] Oauth2 application data for %s: %#v", tokenUrl, string(body))

shared.go

@@ -6692,7 +6692,7 @@ func diffWorkflows(oldWorkflow Workflow, newWorkflow Workflow, update bool) {
 		if err != nil {
 			log.Printf("[WARNING] Failed updating child workflow %s: %s", childWorkflow.ID, err)
 		} else {
-			log.Printf("\n\n[INFO] Updated child workflow '%s' based on parent %s\n\n", childWorkflow.ID, oldWorkflow.ID)
+			log.Printf("[INFO] Updated child workflow '%s' based on parent %s", childWorkflow.ID, oldWorkflow.ID)
 
 			SetWorkflowRevision(ctx, childWorkflow)
 			passedOrg := Org{
@@ -7566,7 +7566,7 @@ func SaveWorkflow(resp http.ResponseWriter, request *http.Request) {
 
 			if added {
 				orgUpdated = true
-				log.Printf("[DEBUG] Org updated with new apps: %s", org.ActiveApps)
+				//log.Printf("[DEBUG] Org updated with new apps: %s", org.ActiveApps)
 
 				//DeleteCache(ctx, fmt.Sprintf("apps_%s", user.Id))
 				DeleteCache(ctx, fmt.Sprintf("workflowapps-sorted-100"))
@@ -15137,7 +15137,8 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut
 				}
 			}
 
-			log.Printf("\n\n[DEBUG][%s] Found that %s (%s) should be skipped? Should check if it has more parents. If not, send in a skip\n\n", workflowExecution.ExecutionId, foundAction.Label, foundAction.AppName)
+			// FIXME: Debug logs necessary to understand how workflows finish?
+			log.Printf("[DEBUG][%s] Found that %s (%s) should be skipped? Should check if it has more parents. If not, send in a skip", workflowExecution.ExecutionId, foundAction.Label, foundAction.AppName)
 
 			foundCount := 0
 			skippedBranches := []string{}
@@ -20885,16 +20886,16 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h
 				setAuth := false
 				executionAuthKey := fmt.Sprintf("oauth2_%s", curAuth.Id)
 
-				log.Printf("[DEBUG] Looking for cached authkey '%s'", executionAuthKey)
+				//log.Printf("[DEBUG] Looking for cached authkey '%s'", executionAuthKey)
 				execAuthData, err := GetCache(ctx, executionAuthKey)
 				if err == nil {
-					log.Printf("[DEBUG] Successfully retrieved auth wrapper from cache for %s", executionAuthKey)
+					//log.Printf("[DEBUG] Successfully retrieved auth wrapper from cache for %s", executionAuthKey)
 					cacheData := []byte(execAuthData.([]uint8))
 
 					appAuthWrapper := AppAuthenticationStorage{}
 					err = json.Unmarshal(cacheData, &appAuthWrapper)
 					if err == nil {
-						log.Printf("[DEBUG] Successfully unmarshalled auth wrapper from cache for %s", executionAuthKey)
+						//log.Printf("[DEBUG] Successfully unmarshalled auth wrapper from cache for %s", executionAuthKey)
 
 						newParams = action.Parameters
 						for _, param := range appAuthWrapper.Fields {