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

#135: Started building app exist validation

Browse Source
This commit is contained in:
frikky
2020-09-05 08:56:29 +02:00
parent b635b6f3e6
commit 2633ff4b4d
8 changed files with 152 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app_sdk/build.sh
+2 -2
View File
@@ -3,11 +3,11 @@ NAME=app_sdk
VERSION=0.6.0 VERSION=0.6.0
docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force
docker build . -t frikky/shuffle:$NAME -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t gchr.io/frikky/app_sdk:0.6.0 docker build . -t frikky/shuffle:$NAME -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/app_sdk:0.6.0
#docker push frikky/shuffle:$NAME #docker push frikky/shuffle:$NAME
#docker push frikky/$NAME:$VERSION #docker push frikky/$NAME:$VERSION
#docker push docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION #docker push docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION
#docker push ghcr.io/frikky/$NAME:$VERSION #docker push ghcr.io/frikky/$NAME:$VERSION
docker push gchr.io/frikky/app_sdk docker push ghcr.io/frikky/$NAME:$VERSION
backend/go-app/docker.go
+43
View File
@@ -642,6 +642,49 @@ func handleStartHookDocker(resp http.ResponseWriter, request *http.Request) {
return return
} }
// Checks if an image exists
func imageCheckBuilder(images []string) error {
log.Printf("ImageNames: %#v", images)
ctx := context.Background()
client, err := client.NewEnvClient()
if err != nil {
log.Printf("Unable to create docker client: %s", err)
return err
}
allImages, err := client.ImageList(ctx, types.ImageListOptions{
All: true,
})
if err != nil {
log.Printf("[ERROR] Failed creating imagelist: %s", err)
return err
}
filteredImages := []types.ImageSummary{}
for _, image := range allImages {
found := false
for _, repoTag := range image.RepoTags {
if strings.Contains(repoTag, baseDockerName) {
found = true
break
}
}
if found {
filteredImages = append(filteredImages, image)
}
}
// FIXME: Continue fixing apps here
// https://github.com/frikky/Shuffle/issues/135
// 1. Find if app exists
// 2. Create app if it doesn't
//log.Printf("Apps: %#v", filteredImages)
return nil
}
func hookTest() { func hookTest() {
var hook Hook var hook Hook
err := json.Unmarshal([]byte(webhook), &hook) err := json.Unmarshal([]byte(webhook), &hook)
backend/go-app/walkoff.go
+20 -32
View File
@@ -42,14 +42,10 @@ var localBase = "http://localhost:5001"
var baseEnvironment = "onprem" var baseEnvironment = "onprem"
var cloudname = "cloud" var cloudname = "cloud"
var defaultLocation = "europe-west2" var defaultLocation = "europe-west2"
var scheduledJobs = map[string]*newscheduler.Job{} var scheduledJobs = map[string]*newscheduler.Job{}
// To test out firestore before potential merge // To test out firestore before potential merge
var shuffleTestProject = "shuffle-test-258209"
var shuffleTestPath = "./shuffle-test-258209-5a2e8d7e508a.json"
//var upgrader = websocket.Upgrader{ //var upgrader = websocket.Upgrader{
// ReadBufferSize: 1024, // ReadBufferSize: 1024,
// WriteBufferSize: 1024, // WriteBufferSize: 1024,
@@ -2146,7 +2142,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
//log.Printf("Execution data: %#v", execution) //log.Printf("Execution data: %#v", execution)
if len(execution.Start) == 36 { if len(execution.Start) == 36 {
log.Printf("SHOULD START ON NODE %s", execution.Start) log.Printf("[INFO] Should start execution on node %s", execution.Start)
workflowExecution.Start = execution.Start workflowExecution.Start = execution.Start
found := false found := false
@@ -2157,12 +2153,12 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
} }
if !found { if !found {
log.Printf("ACTION %s WAS NOT FOUND!", workflow.Start) log.Printf("[ERROR] ACTION %s WAS NOT FOUND!", workflow.Start)
return WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", workflow.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", workflow.Start)) return WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", workflow.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", workflow.Start))
} }
} else if len(execution.Start) > 0 { } else if len(execution.Start) > 0 {
log.Printf("START ACTION %s IS WRONG ID LENGTH %d!", execution.Start, len(execution.Start)) log.Printf("[ERROR] START ACTION %s IS WRONG ID LENGTH %d!", execution.Start, len(execution.Start))
return WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", execution.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", execution.Start)) return WorkflowExecution{}, fmt.Sprintf("Startnode %s was not found in actions", execution.Start), errors.New(fmt.Sprintf("Startnode %s was not found in actions", execution.Start))
} }
@@ -2292,10 +2288,10 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
} }
if len(workflowExecution.ExecutionSource) == 0 { if len(workflowExecution.ExecutionSource) == 0 {
log.Printf("No execution source specified. Setting to default") log.Printf("[INFO] No execution source (trigger) specified. Setting to default")
workflowExecution.ExecutionSource = "default" workflowExecution.ExecutionSource = "default"
} else { } else {
log.Printf("Execution source is %s for execution ID %s", workflowExecution.ExecutionSource, workflowExecution.ExecutionId) log.Printf("[INFO] Execution source is %s for execution ID %s", workflowExecution.ExecutionSource, workflowExecution.ExecutionId)
} }
workflowExecution.ExecutionVariables = workflow.ExecutionVariables workflowExecution.ExecutionVariables = workflow.ExecutionVariables
@@ -2315,7 +2311,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
if len(workflowExecution.Start) == 0 { if len(workflowExecution.Start) == 0 {
workflowExecution.Start = workflowExecution.Workflow.Start workflowExecution.Start = workflowExecution.Workflow.Start
} }
log.Printf("STARTNODE: %s", workflowExecution.Start) log.Printf("[INFO] New startnode: %s", workflowExecution.Start)
childNodes := findChildNodes(workflowExecution, workflowExecution.Start) childNodes := findChildNodes(workflowExecution, workflowExecution.Start)
@@ -2423,6 +2419,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
environments := []string{} environments := []string{}
// Check if the actions are children of the startnode? // Check if the actions are children of the startnode?
imageNames := []string{}
for _, action := range workflowExecution.Workflow.Actions { for _, action := range workflowExecution.Workflow.Actions {
if action.Environment != cloudname { if action.Environment != cloudname {
found := false found := false
@@ -2433,6 +2430,11 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
} }
} }
// Check if the app exists?
newName := action.AppName
newName = strings.ReplaceAll(newName, " ", "-")
imageNames = append(imageNames, fmt.Sprintf("%s:%s_%s", baseDockerName, newName, action.AppVersion))
if !found { if !found {
environments = append(environments, action.Environment) environments = append(environments, action.Environment)
} }
@@ -2441,6 +2443,12 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
} }
} }
err = imageCheckBuilder(imageNames)
if err != nil {
log.Printf("[ERROR] Failed building the required images from %#v: %s", imageNames, err)
return WorkflowExecution{}, "Failed building missing Docker images", err
}
err = setWorkflowExecution(ctx, workflowExecution) err = setWorkflowExecution(ctx, workflowExecution)
if err != nil { if err != nil {
log.Printf("Error saving workflow execution for updates %s: %s", topic, err) log.Printf("Error saving workflow execution for updates %s: %s", topic, err)
@@ -2452,7 +2460,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
if onpremExecution { if onpremExecution {
// FIXME - tmp name based on future companyname-companyId // FIXME - tmp name based on future companyname-companyId
for _, environment := range environments { for _, environment := range environments {
log.Printf("EXECUTION: %s should execute onprem with execution environment \"%s\"", workflowExecution.ExecutionId, environment) log.Printf("[INFO] Execution: %s should execute onprem with execution environment \"%s\"", workflowExecution.ExecutionId, environment)
executionRequest := ExecutionRequest{ executionRequest := ExecutionRequest{
ExecutionId: workflowExecution.ExecutionId, ExecutionId: workflowExecution.ExecutionId,
@@ -2539,7 +2547,7 @@ func executeWorkflow(resp http.ResponseWriter, request *http.Request) {
return return
} }
log.Printf("STARTING EXEC OF %s!", fileId) log.Printf("[INFO] Starting execution of %s!", fileId)
workflowExecution, executionResp, err := handleExecution(fileId, *workflow, request) workflowExecution, executionResp, err := handleExecution(fileId, *workflow, request)
if err == nil { if err == nil {
@@ -3051,26 +3059,6 @@ func getSpecificWorkflow(resp http.ResponseWriter, request *http.Request) {
resp.Write(body) resp.Write(body)
} }
//func setWorkflowExecutionFS(ctx context.Context, reference string, workflowExecution WorkflowExecution) error {
// if len(workflowExecution.ExecutionId) == 0 {
// log.Printf("Workflowexeciton executionId can't be empty.")
// return errors.New("ExecutionId can't be empty.")
// }
//
// firestoreClient, err := firestore.NewClient(ctx, shuffleTestProject, option.WithCredentialsFile(shuffleTestPath))
// if err != nil {
// return err
// }
//
// executionRef := firestoreClient.Doc(reference)
// _, err = executionRef.Set(ctx, workflowExecution)
// if err != nil {
// return err
// }
//
// return nil
//}
func setWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecution) error { func setWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecution) error {
if len(workflowExecution.ExecutionId) == 0 { if len(workflowExecution.ExecutionId) == 0 {
log.Printf("Workflowexeciton executionId can't be empty.") log.Printf("Workflowexeciton executionId can't be empty.")
docker-compose.yml
+6 -4
View File
@@ -2,7 +2,7 @@ version: '3'
services: services:
frontend: frontend:
#build: ./frontend #build: ./frontend
image: frikky/shuffle:frontend image: ghcr.io/frikky/shuffle-frontend:0.7.0
container_name: shuffle-frontend container_name: shuffle-frontend
hostname: shuffle-frontend hostname: shuffle-frontend
ports: ports:
@@ -17,7 +17,7 @@ services:
- backend - backend
backend: backend:
#build: ./backend #build: ./backend
image: frikky/shuffle:backend image: ghcr.io/frikky/shuffle-backend:0.7.0
container_name: shuffle-backend container_name: shuffle-backend
hostname: ${BACKEND_HOSTNAME} hostname: ${BACKEND_HOSTNAME}
# Here for debugging: # Here for debugging:
@@ -43,7 +43,7 @@ services:
- database - database
orborus: orborus:
#build: ./functions/onprem/orborus #build: ./functions/onprem/orborus
image: ghcr.io/frikky/orborus:0.6.0 image: ghcr.io/frikky/orborus:0.6.1
container_name: shuffle-orborus container_name: shuffle-orborus
hostname: shuffle-orborus hostname: shuffle-orborus
networks: networks:
@@ -51,6 +51,8 @@ services:
volumes: volumes:
- /var/run/docker.sock:/var/run/docker.sock - /var/run/docker.sock:/var/run/docker.sock
environment: environment:
- SHUFFLE_APP_SDK_VERSION=0.6.0
- SHUFFLE_WORKER_VERSION=0.6.0
- ORG_ID=${ORG_ID} - ORG_ID=${ORG_ID}
- ENVIRONMENT_NAME=${ENVIRONMENT_NAME} - ENVIRONMENT_NAME=${ENVIRONMENT_NAME}
- BASE_URL=http://${OUTER_HOSTNAME}:${BACKEND_PORT} - BASE_URL=http://${OUTER_HOSTNAME}:${BACKEND_PORT}
@@ -61,7 +63,7 @@ services:
restart: unless-stopped restart: unless-stopped
database: database:
#build: ./backend/database #build: ./backend/database
image: frikky/shuffle:database image: ghcr.io/frikky/shuffle-database:1.0.0
container_name: shuffle-database container_name: shuffle-database
hostname: shuffle-database hostname: shuffle-database
ports: ports:
frontend/src/views/AppCreator.jsx
+1 -1
View File
@@ -1031,7 +1031,7 @@ const AppCreator = (props) => {
/> />
</div> </div>
<div style={{color: "#f85a3e", cursor: "pointer"}} onClick={() => {deletePathQuery(index)}}> <div style={{float: "right", color: "#f85a3e", cursor: "pointer"}} onClick={() => {deletePathQuery(index)}}>
Delete Delete
</div> </div>
functions/onprem/orborus/build.sh
+1 -1
View File
@@ -1,5 +1,5 @@
NAME=orborus NAME=orborus
VERSION=0.6.0 VERSION=0.6.1
echo "Running docker build with $NAME:$VERSION" echo "Running docker build with $NAME:$VERSION"
#docker rmi frikky/shuffle:$NAME --force #docker rmi frikky/shuffle:$NAME --force
functions/onprem/orborus/orborus.go
+50 -39
View File
@@ -30,11 +30,11 @@ var sleepTime = 3
// Timeout if something rashes // Timeout if something rashes
var workerTimeout = 300 var workerTimeout = 300
var appSdkVersion = "0.6.0" var appSdkVersion = os.Getenv("SHUFFLE_APP_SDK_VERSION")
var workerVersion = "0.6.0" var workerVersion = os.Getenv("SHUFFLE_WORKER_VERSION")
//var baseimagename = "docker.pkg.github.com/frikky/shuffle" //var baseimagename = "docker.pkg.github.com/frikky/shuffle"
var baseimagename = "gchr.io/frikky" var baseimagename = "ghcr.io/frikky"
var orgId = os.Getenv("ORG_ID") var orgId = os.Getenv("ORG_ID")
var baseUrl = os.Getenv("BASE_URL") var baseUrl = os.Getenv("BASE_URL")
@@ -121,7 +121,7 @@ func deployWorker(image string, identifier string, env []string) {
log.Printf("[INFO] Found container ID %s", containerId) log.Printf("[INFO] Found container ID %s", containerId)
hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId)) hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
} else { } else {
log.Printf("[WARNING] Empty self container id, continue without NetworkMode") log.Printf("[INFO] Empty self container id, continue without NetworkMode")
} }
config := &container.Config{ config := &container.Config{
@@ -145,7 +145,7 @@ func deployWorker(image string, identifier string, env []string) {
err = dockercli.ContainerStart(context.Background(), cont.ID, types.ContainerStartOptions{}) err = dockercli.ContainerStart(context.Background(), cont.ID, types.ContainerStartOptions{})
if err != nil { if err != nil {
log.Printf("Failed to start container in environment %s: %s", environment, err) log.Printf("[ERROR] Failed to start container in environment %s: %s", environment, err)
return return
//stats, err := cli.ContainerInspect(context.Background(), containerName) //stats, err := cli.ContainerInspect(context.Background(), containerName)
@@ -170,7 +170,7 @@ func deployWorker(image string, identifier string, env []string) {
// } // }
//} //}
} else { } else {
log.Printf("Container %s was created under environment %s", cont.ID, environment) log.Printf("[INFO] Container %s was created under environment %s", cont.ID, environment)
} }
return return
@@ -184,7 +184,7 @@ func stopWorker(containername string) error {
// }) // })
if err := dockercli.ContainerStop(ctx, containername, nil); err != nil { if err := dockercli.ContainerStop(ctx, containername, nil); err != nil {
log.Printf("Unable to stop container %s - running removal anyway, just in case: %s", containername, err) log.Printf("[ERROR] Unable to stop container %s - running removal anyway, just in case: %s", containername, err)
} }
removeOptions := types.ContainerRemoveOptions{ removeOptions := types.ContainerRemoveOptions{
@@ -193,7 +193,7 @@ func stopWorker(containername string) error {
} }
if err := dockercli.ContainerRemove(ctx, containername, removeOptions); err != nil { if err := dockercli.ContainerRemove(ctx, containername, removeOptions); err != nil {
log.Printf("Unable to remove container: %s", err) log.Printf("[ERROR] Unable to remove container: %s", err)
} }
return nil return nil
@@ -202,31 +202,42 @@ func stopWorker(containername string) error {
func initializeImages() { func initializeImages() {
ctx := context.Background() ctx := context.Background()
if appSdkVersion == "" {
appSdkVersion = "0.6.0"
log.Printf("[WARNING] SHUFFLE_APP_SDK_VERSION not defined. Defaulting to %s", appSdkVersion)
}
if workerVersion == "" {
workerVersion = "0.6.0"
log.Printf("[WARNING] SHUFFLE_WORKER_VERSION not defined. Defaulting to %s", workerVersion)
}
// check whether theyre the same first // check whether theyre the same first
images := []string{ images := []string{
// fmt.Sprintf("docker.io/%s:app_sdk", baseimagename), // fmt.Sprintf("docker.io/%s:app_sdk", baseimagename),
// fmt.Sprintf("docker.io/%s:worker", baseimagename), // fmt.Sprintf("docker.io/%s:worker", baseimagename),
fmt.Sprintf("%s/worker:%s", baseimagename, workerVersion), fmt.Sprintf("%s/worker:%s", baseimagename, workerVersion),
fmt.Sprintf("%s/app_sdk:%s", baseimagename, appSdkVersion), fmt.Sprintf("%s/app_sdk:%s", baseimagename, appSdkVersion),
} }
pullOptions := types.ImagePullOptions{} pullOptions := types.ImagePullOptions{}
for _, image := range images { for _, image := range images {
log.Printf("[INFO] Pulling image %s", image)
reader, err := dockercli.ImagePull(ctx, image, pullOptions) reader, err := dockercli.ImagePull(ctx, image, pullOptions)
if err != nil { if err != nil {
log.Printf("Failed getting image %s: %s", image, err) log.Printf("[ERROR] Failed getting image %s: %s", image, err)
continue continue
} }
io.Copy(os.Stdout, reader) io.Copy(os.Stdout, reader)
log.Printf("Successfully downloaded and built %s", image) log.Printf("[INFO] Successfully downloaded and built %s", image)
} }
} }
// Initial loop etc // Initial loop etc
func main() { func main() {
go zombiecheck() go zombiecheck()
log.Println("Setting up execution environment") log.Println("[INFO] Setting up execution environment")
//FIXME //FIXME
if baseUrl == "" { if baseUrl == "" {
@@ -235,30 +246,30 @@ func main() {
} }
if orgId == "" { if orgId == "" {
log.Printf("Org not defined. Set variable ORG_ID based on your org") log.Printf("[ERROR] Org not defined. Set variable ORG_ID based on your org")
os.Exit(3) os.Exit(3)
} }
log.Printf("Running towards %s with Org %s", baseUrl, orgId) log.Printf("[INFO] Running towards %s with Org %s", baseUrl, orgId)
httpProxy := os.Getenv("HTTP_PROXY") httpProxy := os.Getenv("HTTP_PROXY")
httpsProxy := os.Getenv("HTTPS_PROXY") httpsProxy := os.Getenv("HTTPS_PROXY")
if environment == "" { if environment == "" {
environment = "onprem" environment = "onprem"
log.Printf("Defaulting to environment name %s. Set environment variable ENVIRONMENT_NAME to change. This should be the same as in the frontend action.", environment) log.Printf("[INFO] Defaulting to environment name %s. Set environment variable ENVIRONMENT_NAME to change. This should be the same as in the frontend action.", environment)
} }
// FIXME - during init, BUILD and/or LOAD worker and app_sdk // FIXME - during init, BUILD and/or LOAD worker and app_sdk
// Build/load app_sdk so it can be loaded as 127.0.0.1:5000/walkoff_app_sdk // Build/load app_sdk so it can be loaded as 127.0.0.1:5000/walkoff_app_sdk
log.Printf("--- Setting up Docker environment. Downloading worker and App SDK! ---") log.Printf("[INFO] Setting up Docker environment. Downloading worker and App SDK!")
initializeImages() initializeImages()
//workerName := "worker" //workerName := "worker"
//workerVersion := "0.1.0" //workerVersion := "0.1.0"
//workerImage := fmt.Sprintf("docker.pkg.github.com/frikky/shuffle/%s:%s", workerName, workerVersion) //workerImage := fmt.Sprintf("docker.pkg.github.com/frikky/shuffle/%s:%s", workerName, workerVersion)
workerImage := fmt.Sprintf("%s:worker", baseimagename) workerImage := fmt.Sprintf("%s/worker:%s", baseimagename, workerVersion)
log.Printf("--- Finished configuring docker environment ---\n") log.Printf("[INFO] Finished configuring docker environment")
// FIXME - time limit // FIXME - time limit
client := &http.Client{ client := &http.Client{
@@ -271,10 +282,10 @@ func main() {
client = &http.Client{} client = &http.Client{}
} else { } else {
if len(httpProxy) > 0 { if len(httpProxy) > 0 {
log.Printf("Running with HTTP proxy %s (env: HTTP_PROXY)", httpProxy) log.Printf("[INFO] Running with HTTP proxy %s (env: HTTP_PROXY)", httpProxy)
} }
if len(httpsProxy) > 0 { if len(httpsProxy) > 0 {
log.Printf("Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy) log.Printf("[INFO] Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy)
} }
} }
@@ -286,21 +297,21 @@ func main() {
) )
if err != nil { if err != nil {
log.Printf("Failed making request builder: %s", err) log.Printf("[ERROR] Failed making request builder: %s", err)
os.Exit(3) os.Exit(3)
} }
zombiecounter := 0 zombiecounter := 0
req.Header.Add("Content-Type", "application/json") req.Header.Add("Content-Type", "application/json")
req.Header.Add("Org-Id", orgId) req.Header.Add("Org-Id", orgId)
log.Printf("Getting data from %s", fullUrl) log.Printf("[INFO] Waiting for executions at %s", fullUrl)
hasStarted := false hasStarted := false
for { for {
//log.Printf("Prerequest") //log.Printf("Prerequest")
newresp, err := client.Do(req) newresp, err := client.Do(req)
//log.Printf("Postrequest") //log.Printf("Postrequest")
if err != nil { if err != nil {
log.Printf("Failed making request: %s", err) log.Printf("[WARNING] Failed making request: %s", err)
zombiecounter += 1 zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout { if zombiecounter*sleepTime > workerTimeout {
go zombiecheck() go zombiecheck()
@@ -313,7 +324,7 @@ func main() {
// FIXME - add check for StatusCode // FIXME - add check for StatusCode
if newresp.StatusCode != 200 { if newresp.StatusCode != 200 {
if hasStarted { if hasStarted {
log.Printf("Bad statuscode: %d", newresp.StatusCode) log.Printf("[WARNING] Bad statuscode: %d", newresp.StatusCode)
} }
} else { } else {
hasStarted = true hasStarted = true
@@ -321,7 +332,7 @@ func main() {
body, err := ioutil.ReadAll(newresp.Body) body, err := ioutil.ReadAll(newresp.Body)
if err != nil { if err != nil {
log.Printf("Failed reading body: %s", err) log.Printf("[ERROR] Failed reading body: %s", err)
zombiecounter += 1 zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout { if zombiecounter*sleepTime > workerTimeout {
go zombiecheck() go zombiecheck()
@@ -334,7 +345,7 @@ func main() {
var executionRequests ExecutionRequestWrapper var executionRequests ExecutionRequestWrapper
err = json.Unmarshal(body, &executionRequests) err = json.Unmarshal(body, &executionRequests)
if err != nil { if err != nil {
log.Printf("Failed executionrequest in queue unmarshaling: %s", err) log.Printf("[WARNING] Failed executionrequest in queue unmarshaling: %s", err)
sleepTime = 10 sleepTime = 10
zombiecounter += 1 zombiecounter += 1
if zombiecounter*sleepTime > workerTimeout { if zombiecounter*sleepTime > workerTimeout {
@@ -346,7 +357,7 @@ func main() {
} }
if hasStarted && len(executionRequests.Data) > 0 { if hasStarted && len(executionRequests.Data) > 0 {
log.Printf("Body: %s", string(body)) log.Printf("[INFO] Body: %s", string(body))
// Type string `json:"type"` // Type string `json:"type"`
} }
@@ -364,16 +375,16 @@ func main() {
var toBeRemoved ExecutionRequestWrapper var toBeRemoved ExecutionRequestWrapper
for _, execution := range executionRequests.Data { for _, execution := range executionRequests.Data {
if len(execution.ExecutionArgument) > 0 { if len(execution.ExecutionArgument) > 0 {
log.Printf("Argument: %#v", execution.ExecutionArgument) log.Printf("[INFO] Argument: %#v", execution.ExecutionArgument)
} }
if execution.Type == "schedule" { if execution.Type == "schedule" {
log.Printf("SOMETHING ELSE :O: %s", execution.Type) log.Printf("[INFO] SOMETHING ELSE :O: %s", execution.Type)
continue continue
} }
if execution.Status == "ABORT" || execution.Status == "FAILED" { if execution.Status == "ABORT" || execution.Status == "FAILED" {
log.Printf("Executionstatus issue: ", execution.Status) log.Printf("[INFO] Executionstatus issue: ", execution.Status)
} }
// Now, how do I execute this one? // Now, how do I execute this one?
// FIXME - if error, check the status of the running one. If it's bad, send data back. // FIXME - if error, check the status of the running one. If it's bad, send data back.
@@ -396,7 +407,7 @@ func main() {
go deployWorker(workerImage, containerName, env) go deployWorker(workerImage, containerName, env)
log.Printf("%s is deployed and to be removed from queue.", execution.ExecutionId) log.Printf("[INFO] %s is deployed and to be removed from queue.", execution.ExecutionId)
zombiecounter += 1 zombiecounter += 1
toBeRemoved.Data = append(toBeRemoved.Data, execution) toBeRemoved.Data = append(toBeRemoved.Data, execution)
} }
@@ -407,7 +418,7 @@ func main() {
data, err := json.Marshal(toBeRemoved) data, err := json.Marshal(toBeRemoved)
if err != nil { if err != nil {
log.Printf("Failed removal marshalling: %s", err) log.Printf("[WARNING] Failed removal marshalling: %s", err)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
@@ -419,7 +430,7 @@ func main() {
) )
if err != nil { if err != nil {
log.Printf("Failed building confirm request: %s", err) log.Printf("[ERROR] Failed building confirm request: %s", err)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
@@ -429,14 +440,14 @@ func main() {
resultResp, err := client.Do(result) resultResp, err := client.Do(result)
if err != nil { if err != nil {
log.Printf("Failed making confirm request: %s", err) log.Printf("[ERROR] Failed making confirm request: %s", err)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
body, err := ioutil.ReadAll(resultResp.Body) body, err := ioutil.ReadAll(resultResp.Body)
if err != nil { if err != nil {
log.Printf("Failed reading confirm body: %s", err) log.Printf("[ERROR] Failed reading confirm body: %s", err)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
@@ -450,7 +461,7 @@ func main() {
if len(toBeRemoved.Data) == len(executionRequests.Data) { if len(toBeRemoved.Data) == len(executionRequests.Data) {
//log.Println("Should remove ALL!") //log.Println("Should remove ALL!")
} else { } else {
log.Printf("NOT IMPLEMENTED: Should remove %d workflows from backend because they're executed!", len(toBeRemoved.Data)) log.Printf("[INFO] NOT IMPLEMENTED: Should remove %d workflows from backend because they're executed!", len(toBeRemoved.Data))
} }
} }
@@ -461,7 +472,7 @@ func main() {
// FIXME - add this to remove exited workers // FIXME - add this to remove exited workers
// Should it check what happened to the execution? idk // Should it check what happened to the execution? idk
func zombiecheck() error { func zombiecheck() error {
log.Println("Looking for old containers") log.Println("[INFO] Looking for old containers")
ctx := context.Background() ctx := context.Background()
containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{ containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{
@@ -469,7 +480,7 @@ func zombiecheck() error {
}) })
if err != nil { if err != nil {
log.Printf("Failed creating Containerlist: %s", err) log.Printf("[ERROR] Failed creating Containerlist: %s", err)
return err return err
} }
@@ -502,7 +513,7 @@ func zombiecheck() error {
continue continue
} }
log.Printf("NAME: %s", name) log.Printf("[INFO] NAME: %s", name)
// Need to check time here too because a container can be removed the same instant as its created // Need to check time here too because a container can be removed the same instant as its created
currenttime := time.Now().Unix() currenttime := time.Now().Unix()
@@ -522,7 +533,7 @@ func zombiecheck() error {
// FIXME - add killing of apps with same execution ID too // FIXME - add killing of apps with same execution ID too
for _, containername := range stopContainers { for _, containername := range stopContainers {
log.Printf("Stopping and removing container %s", containerNames[containername]) log.Printf("[INFO] Stopping and removing container %s", containerNames[containername])
go dockercli.ContainerStop(ctx, containername, nil) go dockercli.ContainerStop(ctx, containername, nil)
removeContainers = append(removeContainers, containername) removeContainers = append(removeContainers, containername)
} }
functions/onprem/worker/worker.go
+29 -29
View File
@@ -44,7 +44,7 @@ func init() {
if len(containerId) == 0 { if len(containerId) == 0 {
log.Printf("[ERROR] No container ID found.") log.Printf("[ERROR] No container ID found.")
} else { } else {
log.Printf("Found container ID: %s", containerId) log.Printf("[INFO] Found container ID: %s", containerId)
} }
} }
@@ -359,7 +359,7 @@ type ExecutionRequestWrapper struct {
func shutdown(executionId, workflowId string) { func shutdown(executionId, workflowId string) {
dockercli, err := dockerclient.NewEnvClient() dockercli, err := dockerclient.NewEnvClient()
if err != nil { if err != nil {
log.Printf("Unable to create docker client: %s", err) log.Printf("[ERROR] Unable to create docker client: %s", err)
os.Exit(3) os.Exit(3)
} }
@@ -396,7 +396,7 @@ func shutdown(executionId, workflowId string) {
) )
if err != nil { if err != nil {
log.Println("Failed building request: %s", err) log.Println("[INFO] Failed building request: %s", err)
} }
// FIXME: Add an API call to the backend // FIXME: Add an API call to the backend
@@ -404,7 +404,7 @@ func shutdown(executionId, workflowId 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("No authorization specified for abort") log.Printf("[ERROR] No authorization specified for abort")
} }
req.Header.Add("Content-Type", "application/json") req.Header.Add("Content-Type", "application/json")
@@ -420,18 +420,18 @@ func shutdown(executionId, workflowId string) {
client = &http.Client{} client = &http.Client{}
} else { } else {
if len(httpProxy) > 0 { if len(httpProxy) > 0 {
log.Printf("Running with HTTP proxy %s (env: HTTP_PROXY)", httpProxy) log.Printf("[INFO] Running with HTTP proxy %s (env: HTTP_PROXY)", httpProxy)
} }
if len(httpsProxy) > 0 { if len(httpsProxy) > 0 {
log.Printf("Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy) log.Printf("[INFO] Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy)
} }
} }
_, err = client.Do(req) _, err = client.Do(req)
if err != nil { if err != nil {
log.Printf("Failed abort request: %s", err) log.Printf("[INFO] Failed abort request: %s", err)
} }
log.Printf("Finished shutdown.") log.Printf("[INFO] Finished shutdown.")
os.Exit(3) os.Exit(3)
} }
@@ -467,13 +467,12 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
) )
if err != nil { if err != nil {
log.Println(err) log.Printf("Container error: %s", err)
return err return err
} }
cli.ContainerStart(context.Background(), cont.ID, types.ContainerStartOptions{}) cli.ContainerStart(context.Background(), cont.ID, types.ContainerStartOptions{})
fmt.Printf("\n") log.Printf("[INFO] Container %s is created", cont.ID)
log.Printf("Container %s is created", cont.ID)
return nil return nil
} }
@@ -482,7 +481,7 @@ func removeContainer(containername string) error {
cli, err := dockerclient.NewEnvClient() cli, err := dockerclient.NewEnvClient()
if err != nil { if err != nil {
log.Printf("Unable to create docker client: %s", err) log.Printf("[INFO] Unable to create docker client: %s", err)
return err return err
} }
@@ -901,9 +900,11 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W
err = deployApp(dockercli, image, identifier, env) err = deployApp(dockercli, image, identifier, env)
if err != nil { if err != nil {
log.Printf("Failed deploying %s from image %s: %s", identifier, image, err) log.Printf("[ERROR] Failed deploying %s from image %s: %s", identifier, image, err)
log.Printf("Should send status and exit the entire thing?") if strings.Contains(err.Error(), "No such image") {
//shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) log.Printf("[ERROR] Image doesn't exist. Shutting down")
shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID)
}
} }
log.Printf("Adding visited (3): %s", action.Label) log.Printf("Adding visited (3): %s", action.Label)
@@ -1091,7 +1092,7 @@ func runTestExecution(client *http.Client, workflowId, apikey string) (string, s
return "", "" return "", ""
} }
log.Printf("Body: %s", string(body)) log.Printf("[INFO] Body: %s", string(body))
var workflowExecution WorkflowExecution var workflowExecution WorkflowExecution
err = json.Unmarshal(body, &workflowExecution) err = json.Unmarshal(body, &workflowExecution)
if err != nil { if err != nil {
@@ -1104,7 +1105,7 @@ func runTestExecution(client *http.Client, workflowId, apikey string) (string, s
// Initial loop etc // Initial loop etc
func main() { func main() {
log.Printf("Setting up worker environment") log.Printf("[INFO] Setting up worker environment")
sleepTime := 5 sleepTime := 5
client := &http.Client{ client := &http.Client{
@@ -1133,7 +1134,7 @@ func main() {
shuffle_apikey := os.Getenv("WORKER_TESTING_APIKEY") shuffle_apikey := os.Getenv("WORKER_TESTING_APIKEY")
if len(testing) > 0 && len(shuffle_apikey) > 0 { if len(testing) > 0 && len(shuffle_apikey) > 0 {
// Execute a workflow and use that info // Execute a workflow and use that info
log.Printf("!! Running test environment for worker by executing workflow %s", testing) log.Printf("[WARNING] Running test environment for worker by executing workflow %s", testing)
authorization, executionId = runTestExecution(client, testing, shuffle_apikey) authorization, executionId = runTestExecution(client, testing, shuffle_apikey)
//os.Exit(3) //os.Exit(3)
@@ -1144,12 +1145,12 @@ func main() {
} }
if len(authorization) == 0 { if len(authorization) == 0 {
log.Println("No AUTHORIZATION key set in env") log.Println("[INFO] No AUTHORIZATION key set in env")
shutdown(executionId, "") shutdown(executionId, "")
} }
if len(executionId) == 0 { if len(executionId) == 0 {
log.Println("No EXECUTIONID key set in env") log.Println("[INFO] No EXECUTIONID key set in env")
shutdown(executionId, "") shutdown(executionId, "")
} }
@@ -1163,7 +1164,7 @@ func main() {
) )
if err != nil { if err != nil {
log.Println("Failed making request builder") log.Println("[ERROR] Failed making request builder for backend")
shutdown(executionId, "") shutdown(executionId, "")
} }
@@ -1172,20 +1173,20 @@ func main() {
// Removed request requirement from app_sdk // Removed request requirement from app_sdk
newresp, err := client.Do(req) newresp, err := client.Do(req)
if err != nil { if err != nil {
log.Printf("Failed request: %s", err) log.Printf("[ERROR] Failed request: %s", err)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
body, err := ioutil.ReadAll(newresp.Body) body, err := ioutil.ReadAll(newresp.Body)
if err != nil { if err != nil {
log.Printf("Failed reading body: %s", err) log.Printf("[ERROR] Failed reading body: %s", err)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
if newresp.StatusCode != 200 { if newresp.StatusCode != 200 {
log.Printf("Err: %s\nStatusCode: %d", string(body), newresp.StatusCode) log.Printf("[ERROR] %s\nStatusCode: %d", string(body), newresp.StatusCode)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
@@ -1193,13 +1194,13 @@ func main() {
var workflowExecution WorkflowExecution var workflowExecution WorkflowExecution
err = json.Unmarshal(body, &workflowExecution) err = json.Unmarshal(body, &workflowExecution)
if err != nil { if err != nil {
log.Printf("Failed workflowExecution unmarshal: %s", err) log.Printf("[ERROR] Failed workflowExecution unmarshal: %s", err)
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
continue continue
} }
if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" { if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
log.Printf("Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId) log.Printf("[INFO] Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId)
shutdown(executionId, workflowExecution.Workflow.ID) shutdown(executionId, workflowExecution.Workflow.ID)
} }
@@ -1207,15 +1208,14 @@ func main() {
//log.Printf("Status: %s", workflowExecution.Status) //log.Printf("Status: %s", workflowExecution.Status)
err = handleExecution(client, req, workflowExecution) err = handleExecution(client, req, workflowExecution)
if err != nil { if err != nil {
log.Printf("Workflow %s is finished: %s", workflowExecution.ExecutionId, err) log.Printf("[INFO] Workflow %s is finished: %s", workflowExecution.ExecutionId, err)
shutdown(executionId, workflowExecution.Workflow.ID) shutdown(executionId, workflowExecution.Workflow.ID)
} }
} else { } else {
log.Printf("Workflow %s has status %s. Exiting worker.", workflowExecution.ExecutionId, workflowExecution.Status) log.Printf("[INFO] Workflow %s has status %s. Exiting worker.", workflowExecution.ExecutionId, workflowExecution.Status)
shutdown(executionId, workflowExecution.Workflow.ID) shutdown(executionId, workflowExecution.Workflow.ID)
} }
//log.Println(string(body))
time.Sleep(time.Duration(sleepTime) * time.Second) time.Sleep(time.Duration(sleepTime) * time.Second)
} }
} }