marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#415: Added app rebuilds if the container doesn't exist

Browse Source
This commit is contained in:
frikky
2021-06-15 17:41:53 +02:00
parent f970bb351e
commit 642b589953
6 changed files with 280 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/go-app/docker.go
+103 -70
View File
@@ -5,9 +5,9 @@ import (
"github.com/frikky/shuffle-shared" "github.com/frikky/shuffle-shared"
"archive/tar" "archive/tar"
"bufio" //"bufio"
"path/filepath" "path/filepath"
"strconv" //"strconv"
"bytes" "bytes"
"context" "context"
@@ -20,7 +20,6 @@ import (
"github.com/docker/docker/client" "github.com/docker/docker/client"
newdockerclient "github.com/fsouza/go-dockerclient" newdockerclient "github.com/fsouza/go-dockerclient"
"github.com/go-git/go-billy/v5" "github.com/go-git/go-billy/v5"
"github.com/go-git/go-billy/v5/memfs"
//network "github.com/docker/docker/api/types/network" //network "github.com/docker/docker/api/types/network"
//natting "github.com/docker/go-connections/nat" //natting "github.com/docker/go-connections/nat"
@@ -624,13 +623,13 @@ func getDockerImage(resp http.ResponseWriter, request *http.Request) {
} }
// Just here to verify that the user is logged in // Just here to verify that the user is logged in
_, err := shuffle.HandleApiAuthentication(resp, request) //_, err := shuffle.HandleApiAuthentication(resp, request)
if err != nil { //if err != nil {
log.Printf("[WARNING] Api authentication failed in DOWNLOAD IMAGE: %s", err) // log.Printf("[WARNING] Api authentication failed in DOWNLOAD IMAGE: %s", err)
resp.WriteHeader(401) // resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`)) // resp.Write([]byte(`{"success": false}`))
return // return
} //}
body, err := ioutil.ReadAll(request.Body) body, err := ioutil.ReadAll(request.Body)
if err != nil { if err != nil {
@@ -643,16 +642,6 @@ func getDockerImage(resp http.ResponseWriter, request *http.Request) {
Name string `datastore:"name" json:"name" yaml:"name"` Name string `datastore:"name" json:"name" yaml:"name"`
} }
//body = []byte(`swagger: "2.0"`)
//body = []byte(`swagger: '1.0'`)
//newbody := string(body)
//newbody = strings.TrimSpace(newbody)
//body = []byte(newbody)
//log.Println(string(body))
//tmpbody, err := yaml.YAMLToJSON(body)
//log.Println(err)
//log.Println(string(tmpbody))
// This has to be done in a weird way because Datastore doesn't // This has to be done in a weird way because Datastore doesn't
// support map[string]interface and similar (openapi3.Swagger) // support map[string]interface and similar (openapi3.Swagger)
var version requestCheck var version requestCheck
@@ -665,15 +654,9 @@ func getDockerImage(resp http.ResponseWriter, request *http.Request) {
} }
log.Printf("[DEBUG] Image to load: %s", version.Name) log.Printf("[DEBUG] Image to load: %s", version.Name)
//cli, err := client.NewEnvClient()
//if err != nil {
// log.Println("Unable to create docker client")
// return err
//}
dockercli, err := client.NewEnvClient() dockercli, err := client.NewEnvClient()
if err != nil { if err != nil {
log.Printf("Unable to create docker client: %s", err) log.Printf("[WARNING] Unable to create docker client: %s", err)
resp.WriteHeader(422) resp.WriteHeader(422)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed JSON marshalling: %s"}`, err))) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed JSON marshalling: %s"}`, err)))
return return
@@ -686,75 +669,125 @@ func getDockerImage(resp http.ResponseWriter, request *http.Request) {
img := types.ImageSummary{} img := types.ImageSummary{}
tagFound := "" tagFound := ""
img2 := types.ImageSummary{}
tagFound2 := ""
alternativeNameSplit := strings.Split(version.Name, "/")
alternativeName := version.Name
if len(alternativeNameSplit) == 3 {
alternativeName = strings.Join(alternativeNameSplit[1:3], "/")
}
for _, image := range images { for _, image := range images {
for _, tag := range image.RepoTags { for _, tag := range image.RepoTags {
//log.Printf("[INFO] Docker Image: %s", tag)
if strings.ToLower(tag) == strings.ToLower(version.Name) { if strings.ToLower(tag) == strings.ToLower(version.Name) {
img = image img = image
tagFound = tag tagFound = tag
break break
} }
if strings.ToLower(tag) == strings.ToLower(alternativeName) {
img2 = image
tagFound2 = tag
}
} }
} }
// REBUILDS THE APP
if len(img.ID) == 0 { if len(img.ID) == 0 {
resp.WriteHeader(401) if len(img2.ID) == 0 {
resp.Write([]byte(fmt.Sprintf(`{"success": false, "message": "Couldn't find image %s"}`, version.Name))) workflowapps, err := shuffle.GetAllWorkflowApps(ctx, 500)
return log.Printf("[INFO] Getting workflowapps for a rebuild. Got %d with err %#v", len(workflowapps), err)
if err == nil {
imageName := ""
imageVersion := ""
newNameSplit := strings.Split(version.Name, ":")
if len(newNameSplit) == 2 {
log.Printf("[DEBUG] Found name %#v", newNameSplit)
findVersionSplit := strings.Split(newNameSplit[1], "_")
log.Printf("[DEBUG] Found another split %#v", findVersionSplit)
if len(findVersionSplit) == 2 {
imageVersion = findVersionSplit[len(findVersionSplit)-1]
imageName = findVersionSplit[0]
} else if len(findVersionSplit) >= 2 {
imageVersion = findVersionSplit[len(findVersionSplit)-1]
imageName = strings.Join(findVersionSplit[0:len(findVersionSplit)-1], "_")
} else {
log.Printf("[DEBUG] Couldn't parse appname & version for %#v", findVersionSplit)
}
}
if len(imageName) > 0 && len(imageVersion) > 0 {
log.Printf("Looking for appname %s with version %s", imageName, imageVersion)
foundApp := shuffle.WorkflowApp{}
for _, app := range workflowapps {
if strings.ToLower(strings.Replace(app.Name, " ", "_", -1)) == imageName && app.AppVersion == imageVersion {
if app.Generated {
foundApp = app
break
}
break
}
}
if len(foundApp.ID) > 0 {
openApiApp, err := shuffle.GetOpenApiDatastore(ctx, foundApp.ID)
if err != nil {
log.Printf("[ERROR] Failed getting OpenAPI app %s to database: %s", foundApp.ID, err)
} else {
log.Printf("[DEBUG] Found OpenAPI app for %s as generated - now building!", version.Name)
user := shuffle.User{}
//img = version.Name
if len(alternativeName) > 0 {
tagFound = alternativeName
} else {
tagFound = version.Name
}
buildSwaggerApp(resp, []byte(openApiApp.Body), user)
}
}
}
} else {
log.Printf("[WARNING] Couldn't find an image with registry name %s and %s", version.Name, alternativeName)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "message": "Couldn't find image %s"}`, version.Name)))
return
}
}
if len(tagFound) == 0 && len(tagFound2) > 0 {
img = img2
tagFound = tagFound2
}
} }
log.Printf("[INFO] Img found (%s): %#v", tagFound, img) //log.Printf("[INFO] Img found (%s): %#v", tagFound, img)
log.Printf("[INFO] Img found to be downloaded: %s", tagFound)
basepath := "base"
location := fmt.Sprintf("%s.tar.gz", tagFound)
fs := memfs.New()
//Close after function return
f, err := fs.Create(fmt.Sprintf("%s/%s", basepath, location))
if err != nil {
log.Printf("[WARNING] Failed making file: %s", err)
return
}
newClient, err := newdockerclient.NewClientFromEnv() newClient, err := newdockerclient.NewClientFromEnv()
if err != nil { if err != nil {
log.Printf("[WARNING] Failed setting up docker env: %s", newClient) log.Printf("[WARNING] Failed setting up docker env: %s", newClient)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "message": "Couldn't make docker client"}`)))
return return
} }
//https://github.com/fsouza/go-dockerclient/issues/600 ////https://github.com/fsouza/go-dockerclient/issues/600
defer f.Close() //defer fileReader.Close()
w := bufio.NewWriter(f)
opts := newdockerclient.ExportImageOptions{ opts := newdockerclient.ExportImageOptions{
Name: tagFound, Name: tagFound,
OutputStream: w, OutputStream: resp,
} }
if err := newClient.ExportImage(opts); err != nil { if err := newClient.ExportImage(opts); err != nil {
log.Printf("[WARNING] FAILED to save image to file: %s", err) log.Printf("[WARNING] FAILED to save image to file: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "message": "Couldn't export image"}`)))
return return
} }
w.Flush()
FileHeader := make([]byte, 512)
f.Read(FileHeader)
FileContentType := http.DetectContentType(FileHeader)
//Get the file size
//FileStat, _ := f.Stat() //Get info from file
//FileSize := strconv.FormatInt(f.Size(), 10) //Get file size as a string
//Send the headers
resp.Header().Set("Content-Disposition", "attachment; filename="+location)
resp.Header().Set("Content-Type", FileContentType)
resp.Header().Set("Content-Length", strconv.FormatInt(img.Size, 10))
//Send the file
//We read 512 bytes from the file already, so we reset the offset back to 0
f.Seek(0, 0)
io.Copy(resp, f) //'Copy' the file to the client
//resp.WriteHeader(200)
//resp.Write([]byte(fmt.Sprintf(`{"success": true, "message": "Downloading image %s"}`, version.Name)))
} }
backend/go-app/go.sum
+2
View File
@@ -177,6 +177,8 @@ github.com/frikky/shuffle-shared v0.0.60 h1:o6/QLsu3Rbjr4+BQWs9DF4B5qZCg0gPpZWPY
github.com/frikky/shuffle-shared v0.0.60/go.mod h1:BknTfpun3qte5bumR3OqQHf9XWPIsyj8woiXCjIlbBc= github.com/frikky/shuffle-shared v0.0.60/go.mod h1:BknTfpun3qte5bumR3OqQHf9XWPIsyj8woiXCjIlbBc=
github.com/frikky/shuffle-shared v0.0.62 h1:1M8y7rX8nQW7072+bUD4vgHcf65AG0kJ8m3ihmY2bPQ= github.com/frikky/shuffle-shared v0.0.62 h1:1M8y7rX8nQW7072+bUD4vgHcf65AG0kJ8m3ihmY2bPQ=
github.com/frikky/shuffle-shared v0.0.62/go.mod h1:BknTfpun3qte5bumR3OqQHf9XWPIsyj8woiXCjIlbBc= github.com/frikky/shuffle-shared v0.0.62/go.mod h1:BknTfpun3qte5bumR3OqQHf9XWPIsyj8woiXCjIlbBc=
github.com/frikky/shuffle-shared v0.0.63 h1:btn7V7s98eZmx/9qyapGE3V1TyxftvjMmrzkEcUvu+c=
github.com/frikky/shuffle-shared v0.0.63/go.mod h1:oPDyGFBuurPtE6UQLUrHU/JtbOJftqDljRfEdLlCTI4=
github.com/fsouza/go-dockerclient v1.7.2 h1:bBEAcqLTkpq205jooP5RVroUKiVEWgGecHyeZc4OFjo= github.com/fsouza/go-dockerclient v1.7.2 h1:bBEAcqLTkpq205jooP5RVroUKiVEWgGecHyeZc4OFjo=
github.com/fsouza/go-dockerclient v1.7.2/go.mod h1:+ugtMCVRwnPfY7d8/baCzZ3uwB0BrG5DB8OzbtxaRz8= github.com/fsouza/go-dockerclient v1.7.2/go.mod h1:+ugtMCVRwnPfY7d8/baCzZ3uwB0BrG5DB8OzbtxaRz8=
github.com/getkin/kin-openapi v0.8.0 h1:a6TQjTqwkyscC4/hShJX7WhCVE+4bi9lzw61XHQW5hE= github.com/getkin/kin-openapi v0.8.0 h1:a6TQjTqwkyscC4/hShJX7WhCVE+4bi9lzw61XHQW5hE=
backend/go-app/main.go
+42 -66
View File
@@ -3081,29 +3081,7 @@ func handleSwaggerValidation(body []byte) (shuffle.ParsedOpenApi, error) {
return parsed, err return parsed, err
} }
// Creates an app from the app builder func buildSwaggerApp(resp http.ResponseWriter, body []byte, user shuffle.User) {
func verifySwagger(resp http.ResponseWriter, request *http.Request) {
cors := handleCors(resp, request)
if cors {
return
}
//log.Printf("[INFO] TRY TO SET APP TO LIVE!!!")
user, err := shuffle.HandleApiAuthentication(resp, request)
if err != nil {
log.Printf("Api authentication failed in verify swagger: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
}
body, err := ioutil.ReadAll(request.Body)
if err != nil {
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false, "reason": "Failed reading body"}`))
return
}
type Test struct { type Test struct {
Editing bool `datastore:"editing"` Editing bool `datastore:"editing"`
Id string `datastore:"id"` Id string `datastore:"id"`
@@ -3111,7 +3089,7 @@ func verifySwagger(resp http.ResponseWriter, request *http.Request) {
} }
var test Test var test Test
err = json.Unmarshal(body, &test) err := json.Unmarshal(body, &test)
if err != nil { if err != nil {
log.Printf("Failed unmarshalling test: %s", err) log.Printf("Failed unmarshalling test: %s", err)
resp.WriteHeader(401) resp.WriteHeader(401)
@@ -3123,7 +3101,8 @@ func verifySwagger(resp http.ResponseWriter, request *http.Request) {
hasher := md5.New() hasher := md5.New()
hasher.Write(body) hasher.Write(body)
newmd5 := hex.EncodeToString(hasher.Sum(nil)) newmd5 := hex.EncodeToString(hasher.Sum(nil))
if test.Editing {
if test.Editing && len(user.Id) > 0 {
// Quick verification test // Quick verification test
ctx := context.Background() ctx := context.Background()
app, err := shuffle.GetApp(ctx, test.Id, user) app, err := shuffle.GetApp(ctx, test.Id, user)
@@ -3150,25 +3129,6 @@ func verifySwagger(resp http.ResponseWriter, request *http.Request) {
// Test = client side with fetch? // Test = client side with fetch?
ctx := context.Background() ctx := context.Background()
//s := string(body)
//if !utf8.ValidString(s) {
// v := make([]rune, 0, len(s))
// for i, r := range s {
// if r == utf8.RuneError {
// _, size := utf8.DecodeRuneInString(s[i:])
// if size == 1 {
// continue
// }
// }
// v = append(v, r)
// }
// s = string(v)
//}
//fmt.Printf("%q\n", s)
//body = []byte(strings.Replace(string(body), '<80>', '', -1))
//log.Println(string(body))
swaggerLoader := openapi3.NewSwaggerLoader() swaggerLoader := openapi3.NewSwaggerLoader()
swaggerLoader.IsExternalRefsAllowed = true swaggerLoader.IsExternalRefsAllowed = true
swagger, err := swaggerLoader.LoadSwaggerFromData(body) swagger, err := swaggerLoader.LoadSwaggerFromData(body)
@@ -3300,20 +3260,6 @@ func verifySwagger(resp http.ResponseWriter, request *http.Request) {
log.Printf("[INFO] Successfully stitched ZIPFILE for %s", identifier) log.Printf("[INFO] Successfully stitched ZIPFILE for %s", identifier)
// 4. Upload as cloud function - this apikey is specifically for cloud functions rofl
//environmentVariables := map[string]string{
// "FUNCTION_APIKEY": apikey,
//}
//fullLocation := fmt.Sprintf("gs://%s/%s", bucketName, applocation)
//err = deployCloudFunctionPython(ctx, identifier, defaultLocation, fullLocation, environmentVariables)
//if err != nil {
// log.Printf("Error uploading cloud function: %s", err)
// resp.WriteHeader(500)
// resp.Write([]byte(`{"success": false, "reason": "Failed to upload function"}`))
// return
//}
// 4. Build the image locally. // 4. Build the image locally.
// FIXME: Should be moved to a local docker registry // FIXME: Should be moved to a local docker registry
dockerLocation := fmt.Sprintf("%s/Dockerfile", basePath) dockerLocation := fmt.Sprintf("%s/Dockerfile", basePath)
@@ -3359,12 +3305,14 @@ func verifySwagger(resp http.ResponseWriter, request *http.Request) {
user.PrivateApps[foundNumber] = api user.PrivateApps[foundNumber] = api
} }
err = shuffle.SetUser(ctx, &user, true) if len(user.Id) > 0 {
if err != nil { err = shuffle.SetUser(ctx, &user, true)
log.Printf("[ERROR] Failed adding verification for user %s: %s", user.Username, err) if err != nil {
resp.WriteHeader(500) log.Printf("[ERROR] Failed adding verification for user %s: %s", user.Username, err)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Failed updating user"}`))) resp.WriteHeader(500)
return resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Failed updating user"}`)))
return
}
} }
//log.Printf("DO I REACH HERE WHEN SAVING?") //log.Printf("DO I REACH HERE WHEN SAVING?")
@@ -3405,8 +3353,36 @@ func verifySwagger(resp http.ResponseWriter, request *http.Request) {
shuffle.DeleteCache(ctx, cacheKey) shuffle.DeleteCache(ctx, cacheKey)
shuffle.DeleteCache(ctx, fmt.Sprintf("apps_%s", user.Id)) shuffle.DeleteCache(ctx, fmt.Sprintf("apps_%s", user.Id))
resp.WriteHeader(200) if len(user.Id) > 0 {
resp.Write([]byte(fmt.Sprintf(`{"success": true, "id": "%s"}`, api.ID))) resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "id": "%s"}`, api.ID)))
}
}
// Creates an app from the app builder
func verifySwagger(resp http.ResponseWriter, request *http.Request) {
cors := handleCors(resp, request)
if cors {
return
}
//log.Printf("[INFO] TRY TO SET APP TO LIVE!!!")
user, err := shuffle.HandleApiAuthentication(resp, request)
if err != nil {
log.Printf("Api authentication failed in verify swagger: %s", err)
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false}`))
return
}
body, err := ioutil.ReadAll(request.Body)
if err != nil {
resp.WriteHeader(401)
resp.Write([]byte(`{"success": false, "reason": "Failed reading body"}`))
return
}
buildSwaggerApp(resp, body, user)
} }
func healthCheckHandler(resp http.ResponseWriter, request *http.Request) { func healthCheckHandler(resp http.ResponseWriter, request *http.Request) {
backend/tests/dockerpull.sh
+4 -1
View File
@@ -1,2 +1,5 @@
#!/bin/sh #!/bin/sh
curl -XPOST http://localhost:5001/api/v1/get_docker_image -d '{"name":"frikky/shuffle:Testing_1.0.0"}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" --output tarball.tgz #curl -XPOST http://localhost:5001/api/v1/get_docker_image -d '{"name":"frikky/shuffle:testing_1.0.0"}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" --output tarball.tgz
#curl -XPOST http://localhost:5001/api/v1/get_docker_image -d '{"name":"frikky/shuffle:testing_1.0.0"}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" -O -J
#curl -XPOST http://localhost:5001/api/v1/get_docker_image -d '{"name":"frikky/shuffle:testing_1.0.0"}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" --output tarball.tgz
curl -XPOST http://localhost:5001/api/v1/get_docker_image -d '{"name":"frikky/shuffle:shuffle-tools_1.0.0"}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" --output tarball.tgz
functions/onprem/worker/build.sh
+1 -1
View File
@@ -1,5 +1,5 @@
NAME=shuffle-worker NAME=shuffle-worker
VERSION=0.8.100 VERSION=0.8.101
echo "Running docker build with $NAME:$VERSION" echo "Running docker build with $NAME:$VERSION"
#CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin . #CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin .
functions/onprem/worker/worker.go
+128 -120
View File
@@ -27,7 +27,7 @@ import (
dockerclient "github.com/docker/docker/client" dockerclient "github.com/docker/docker/client"
//"github.com/go-git/go-billy/v5/memfs" //"github.com/go-git/go-billy/v5/memfs"
newdockerclient "github.com/fsouza/go-dockerclient" //newdockerclient "github.com/fsouza/go-dockerclient"
//"github.com/satori/go.uuid" //"github.com/satori/go.uuid"
"github.com/gorilla/mux" "github.com/gorilla/mux"
@@ -58,6 +58,7 @@ var containerIds []string
var extra int var extra int
var startAction string var startAction string
var results []shuffle.ActionResult var results []shuffle.ActionResult
var allLogs map[string]string
var containerId string var containerId string
@@ -93,7 +94,7 @@ func init() {
// removes every container except itself (worker) // removes every container except itself (worker)
func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) { func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) {
log.Printf("[INFO] Shutdown (%s) started with reason %s. Result amount: %d. ResultsSent: %d, Send result: %#v", workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend) log.Printf("[INFO] Shutdown (%s) started with reason %#v. Result amount: %d. ResultsSent: %d, Send result: %#v", workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend)
//reason := "Error in execution" //reason := "Error in execution"
sleepDuration := 1 sleepDuration := 1
@@ -184,19 +185,23 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
log.Printf("[INFO] Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy) log.Printf("[INFO] Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy)
} }
} }
log.Printf("[INFO] All App Logs: %#v", allLogs)
_, err = client.Do(req) _, err = client.Do(req)
if err != nil { if err != nil {
log.Printf("[INFO] Failed abort request: %s", err) log.Printf("[WARNING] Failed abort request: %s", err)
} }
log.Printf("[INFO] Finished shutdown (after %d seconds).", sleepDuration) log.Printf("[INFO] Finished shutdown (after %d seconds). ", sleepDuration)
// Allows everything to finish in subprocesses //Finished shutdown (after %d seconds). ", sleepDuration)
// Allows everything to finish in subprocesses (apps)
time.Sleep(time.Duration(sleepDuration) * time.Second) time.Sleep(time.Duration(sleepDuration) * time.Second)
os.Exit(3) os.Exit(3)
} }
// Deploys the internal worker whenever something happens // Deploys the internal worker whenever something happens
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution) error { func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, actionId string) error {
// form basic hostConfig // form basic hostConfig
ctx := context.Background() ctx := context.Background()
@@ -289,7 +294,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
// Waiting to see if it exits.. Stupid, but stable(r) // Waiting to see if it exits.. Stupid, but stable(r)
if workflowExecution.ExecutionSource != "default" { if workflowExecution.ExecutionSource != "default" {
log.Printf("[INFO] Handling NON-default execution source %s - NOT waiting and validating!", workflowExecution.ExecutionSource) log.Printf("[INFO] Handling NON-default execution source %s - NOT waiting or validating!", workflowExecution.ExecutionSource)
} else if workflowExecution.ExecutionSource == "default" { } else if workflowExecution.ExecutionSource == "default" {
time.Sleep(2 * time.Second) time.Sleep(2 * time.Second)
@@ -301,30 +306,29 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
//log.Printf("%#v", stats.Config) //log.Printf("%#v", stats.Config)
//log.Printf("%#v", stats.ContainerJSONBase.State) //log.Printf("%#v", stats.ContainerJSONBase.State)
log.Printf("[INFO] EXECUTION STATUS: %s", stats.ContainerJSONBase.State.Status) log.Printf("[INFO] EXECUTION STATUS: %s", stats.ContainerJSONBase.State.Status)
if stats.ContainerJSONBase.State.Status == "exited" { logOptions := types.ContainerLogsOptions{
logOptions := types.ContainerLogsOptions{ ShowStdout: true,
ShowStdout: true, }
}
exit := true exit := true
out, err := cli.ContainerLogs(ctx, cont.ID, logOptions) out, err := cli.ContainerLogs(ctx, cont.ID, logOptions)
if err != nil { if err != nil {
log.Printf("[INFO] Failed getting logs: %s", err) log.Printf("[INFO] Failed getting logs: %s", err)
} else { } else {
buf := new(strings.Builder) buf := new(strings.Builder)
io.Copy(buf, out) io.Copy(buf, out)
logs := buf.String() logs := buf.String()
log.Printf("Execution Logs: %s", logs) //allLogs[actionId] = logs
if strings.Contains(logs, "Normal execution.") { if stats.ContainerJSONBase.State.Status == "exited" && strings.Contains(logs, "Normal execution.") {
exit = false log.Printf("[WARNING] BAD Execution Logs for %s: %s", actionId, logs)
} exit = false
} }
}
if exit { if exit {
log.Printf("ERROR IN CONTAINER DEPLOYMENT - ITS EXITED!") log.Printf("ERROR IN CONTAINER DEPLOYMENT - ITS EXITED!")
return errors.New(fmt.Sprintf(`{"success": false, "reason": "Container %s exited prematurely.","debug": "docker logs -f %s"}`, cont.ID, cont.ID)) return errors.New(fmt.Sprintf(`{"success": false, "reason": "Container %s exited prematurely.","debug": "docker logs -f %s"}`, cont.ID, cont.ID))
}
} }
} }
} }
@@ -348,7 +352,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
logOptions := types.ContainerLogsOptions{ logOptions := types.ContainerLogsOptions{
ShowStdout: true, ShowStdout: true,
} }
*/ */
containerIds = append(containerIds, cont.ID) containerIds = append(containerIds, cont.ID)
@@ -443,7 +446,7 @@ func handleSubworkflowExecution(client *http.Client, workflowExecution shuffle.W
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", apikey)) req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", apikey))
newresp, err := client.Do(req) newresp, err := client.Do(req)
if err != nil { if err != nil {
log.Printf("Error running test request: %s", err) log.Printf("[DEBUG] Error running test request: %s", err)
return err return err
} }
@@ -984,7 +987,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
// If cleanup is set, it should run for efficiency // If cleanup is set, it should run for efficiency
pullOptions := types.ImagePullOptions{} pullOptions := types.ImagePullOptions{}
if cleanupEnv == "true" { if cleanupEnv == "true" {
err = deployApp(dockercli, images[0], identifier, env, workflowExecution) err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action.ID)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") { if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (2)") log.Printf("[DEBUG] Shutting down (2)")
@@ -992,37 +995,46 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
} }
image = images[2] image = images[2]
err = deployApp(dockercli, image, identifier, env, workflowExecution) err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") { if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (3)") log.Printf("[DEBUG] Shutting down (3)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true) shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} }
log.Printf("[WARNING] Failed CLEANUP execution. Downloading image remotely.")
reader, err := dockercli.ImagePull(context.Background(), image, pullOptions)
if err != nil {
log.Printf("[ERROR] Failed getting %s. Couldn't be find locally, AND is missing.", image)
log.Printf("[DEBUG] Shutting down (4)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
}
buildBuf := new(strings.Builder) log.Printf("[WARNING] Failed CLEANUP execution. Downloading image %s remotely.", image)
_, err = io.Copy(buildBuf, reader)
if err != nil && !strings.Contains(fmt.Sprintf("%s", err.Error()), "Conflict. The container name") { err := downloadDockerImageBackend(topClient, image)
log.Printf("[ERROR] Error in IO copy: %s", err) if err == nil {
log.Printf("[DEBUG] Shutting down (5)") log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} else { } else {
if strings.Contains(buildBuf.String(), "errorDetail") { log.Printf("[WARNING] Failed to download image %s (CLEANUP): %s", image, err)
log.Printf("[ERROR] Docker build:\n%s\nERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), image)
log.Printf("[DEBUG] Shutting down (6)") reader, err := dockercli.ImagePull(context.Background(), image, pullOptions)
if err != nil {
log.Printf("[ERROR] Failed getting %s. Couldn't be find locally, AND is missing.", image)
log.Printf("[DEBUG] Shutting down (4)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true) shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} }
log.Printf("[INFO] Successfully downloaded %s", image) buildBuf := new(strings.Builder)
_, err = io.Copy(buildBuf, reader)
if err != nil && !strings.Contains(fmt.Sprintf("%s", err.Error()), "Conflict. The container name") {
log.Printf("[ERROR] Error in IO copy: %s", err)
log.Printf("[DEBUG] Shutting down (5)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} else {
if strings.Contains(buildBuf.String(), "errorDetail") {
log.Printf("[ERROR] Docker build:\n%s\nERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), image)
log.Printf("[DEBUG] Shutting down (6)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
}
log.Printf("[INFO] Successfully downloaded %s", image)
}
} }
err = deployApp(dockercli, image, identifier, env, workflowExecution) err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist") log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
@@ -1042,7 +1054,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
} }
} else { } else {
err = deployApp(dockercli, images[0], identifier, env, workflowExecution) err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action.ID)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") { if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (9)") log.Printf("[DEBUG] Shutting down (9)")
@@ -1052,7 +1064,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
// Trying to replace with lowercase to deploy again. This seems to work with Dockerhub well. // Trying to replace with lowercase to deploy again. This seems to work with Dockerhub well.
// FIXME: Should try to remotely download directly if this persists. // FIXME: Should try to remotely download directly if this persists.
image = images[1] image = images[1]
err = deployApp(dockercli, image, identifier, env, workflowExecution) err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") { if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (10)") log.Printf("[DEBUG] Shutting down (10)")
@@ -1060,38 +1072,44 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
} }
image = images[2] image = images[2]
err = deployApp(dockercli, image, identifier, env, workflowExecution) err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") { if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (11)") log.Printf("[DEBUG] Shutting down (11)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true) shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} }
log.Printf("[WARNING] Failed deploying image THREE TIMES. Attempting to download the latter as last resort.") log.Printf("[WARNING] Failed deploying image THREE TIMES. Attempting to download %s as last resort from backend and dockerhub.", image)
reader, err := dockercli.ImagePull(context.Background(), image, pullOptions)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed getting %s. The couldn't be find locally, AND is missing.", image)
log.Printf("[DEBUG] Shutting down (12)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
}
buildBuf := new(strings.Builder) err := downloadDockerImageBackend(topClient, image)
_, err = io.Copy(buildBuf, reader) if err == nil {
if err != nil { log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
log.Printf("[ERROR] Error in IO copy: %s", err)
log.Printf("[DEBUG] Shutting down (13)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} else { } else {
if strings.Contains(buildBuf.String(), "errorDetail") { reader, err := dockercli.ImagePull(context.Background(), image, pullOptions)
log.Printf("[ERROR] Docker build:\n%s\nERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), image) if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[DEBUG] Shutting down (14)") log.Printf("[ERROR] Failed getting %s. The couldn't be find locally, AND is missing.", image)
log.Printf("[DEBUG] Shutting down (12)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true) shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} }
log.Printf("[INFO] Successfully downloaded %s", image) buildBuf := new(strings.Builder)
_, err = io.Copy(buildBuf, reader)
if err != nil {
log.Printf("[ERROR] Error in IO copy: %s", err)
log.Printf("[DEBUG] Shutting down (13)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
} else {
if strings.Contains(buildBuf.String(), "errorDetail") {
log.Printf("[ERROR] Docker build:\n%s\nERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), image)
log.Printf("[DEBUG] Shutting down (14)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
}
log.Printf("[INFO] Successfully downloaded %s", image)
}
} }
err = deployApp(dockercli, image, identifier, env, workflowExecution) err = deployApp(dockercli, image, identifier, env, workflowExecution, action.ID)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") { if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist") log.Printf("[ERROR] Failed deploying image for the FOURTH time. Aborting if the image doesn't exist")
if strings.Contains(err.Error(), "exited prematurely") { if strings.Contains(err.Error(), "exited prematurely") {
@@ -1556,7 +1574,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
if err != nil { if err != nil {
log.Printf("[ERROR] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, err) log.Printf("[ERROR] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, err)
resp.WriteHeader(401) resp.WriteHeader(401)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist."}`, actionResult.ExecutionId))) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist locally."}`, actionResult.ExecutionId)))
return return
} }
@@ -1895,7 +1913,8 @@ func runWebserver(listener net.Listener) {
log.Fatal(http.Serve(listener, nil)) log.Fatal(http.Serve(listener, nil))
} }
func downloadDockerImage(client *http.Client, imageName string) { func downloadDockerImageBackend(client *http.Client, imageName string) error {
log.Printf("[DEBUG] Trying to download image %s from backend as it doesn't exist", imageName)
data := fmt.Sprintf(`{"name": "%s"}`, imageName) data := fmt.Sprintf(`{"name": "%s"}`, imageName)
dockerImgUrl := fmt.Sprintf("%s/api/v1/get_docker_image", baseUrl) dockerImgUrl := fmt.Sprintf("%s/api/v1/get_docker_image", baseUrl)
@@ -1909,75 +1928,65 @@ func downloadDockerImage(client *http.Client, imageName string) {
if len(authorization) > 0 { if len(authorization) > 0 {
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization)) req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization))
} else { } else {
log.Printf("[WARNING] No auth found.") log.Printf("[WARNING] No auth found - running backend download without it.")
req.Header.Add("Authorization", fmt.Sprintf("Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4")) //req.Header.Add("Authorization", fmt.Sprintf("Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4"))
//return //return
} }
newresp, err := client.Do(req) newresp, err := client.Do(req)
if err != nil { if err != nil {
log.Printf("[ERROR] Failed request: %s", err) log.Printf("[ERROR] Failed request: %s", err)
return return err
} }
if newresp.StatusCode != 200 { if newresp.StatusCode != 200 {
log.Printf("[ERROR] DOWNLOAD StatusCode (1): %d", newresp.StatusCode) log.Printf("[ERROR] Docker download for image %s (backend) StatusCode (1): %d", imageName, newresp.StatusCode)
return return errors.New(fmt.Sprintf("Failed to get image - status code %d", newresp.StatusCode))
}
// Write the body to file
newClient, err := newdockerclient.NewClientFromEnv()
if err != nil {
log.Printf("[WARNING] Failed setting up docker env in download: %s", newClient)
return
} }
newImageName := strings.Replace(imageName, "/", "_", -1) newImageName := strings.Replace(imageName, "/", "_", -1)
newFileName := newImageName + ".tar.gz" newFileName := newImageName + ".tar"
//os.Create(newFileName)
tar, err := os.Create(newFileName) tar, err := os.Create(newFileName)
if err != nil { if err != nil {
log.Printf("[WARNING] Failed creating file: %s", err) log.Printf("[WARNING] Failed creating file: %s", err)
return return err
} }
//fs := memfs.New()
//if err != nil {
// log.Printf("[WARNING] Failed making memory file: %s", err)
// return
//}
//imageName = strings.Replace(imageName, "/", "_", -1)
//tar, err := fs.Create(imageName + ".tar.gz")
//if err != nil {
// log.Printf("[WARNING] Failed making file: %s", err)
// return
//}
defer tar.Close() defer tar.Close()
_, err = io.Copy(tar, newresp.Body) _, err = io.Copy(tar, newresp.Body)
//OutputStream: outFile,
//Context: context.Background(),
imageOptions := newdockerclient.LoadImageOptions{
InputStream: tar,
}
//log.Printf("BUF: %s", buf.String())
err = newClient.LoadImage(imageOptions)
if err != nil { if err != nil {
log.Printf("[WARNING] Failed loading image %s: %s", imageName, err) log.Printf("[WARNING] Failed response body copying: %s", err)
return return err
}
tar.Seek(0, 0)
dockercli, err := dockerclient.NewEnvClient()
if err != nil {
log.Printf("[ERROR] Unable to create docker client (3): %s", err)
return err
} }
log.Printf("[INFO] Successfully loaded image %s", imageName) imageLoadResponse, err := dockercli.ImageLoad(context.Background(), tar, true)
//err = os.Remove(newImageName) if err != nil {
//if err != nil { log.Printf("[ERROR] Error loading: %s", err)
// log.Printf("[WARNING] Failed removing file: %s", err) return err
//} }
return 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))
}
os.Remove(newFileName)
log.Printf("[INFO] Successfully loaded image %s: %s", imageName, string(body))
return nil
} }
// Initial loop etc // Initial loop etc
@@ -2004,9 +2013,8 @@ func main() {
} }
} }
//imageName := "frikky/shuffle:Testing_1.0.0" //imageName := fmt.Sprintf("%s/%s:shuffle_openapi_1.0.0", registryName, baseimagename)
//downloadDockerImage(client, imageName) //downloadDockerImageBackend(client, imageName)
//os.Exit(3)
// WORKER_TESTING_WORKFLOW should be a workflow ID // WORKER_TESTING_WORKFLOW should be a workflow ID
authorization := "" authorization := ""