diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index c8acad84..8f28f72e 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -17,6 +17,9 @@ import ( "strings" "time" + "github.com/docker/docker/api/types" + "github.com/docker/docker/client" + "cloud.google.com/go/datastore" scheduler "cloud.google.com/go/scheduler/apiv1" gyaml "github.com/ghodss/yaml" @@ -977,6 +980,21 @@ func validateNewWorkerExecution(body []byte) error { return errors.New(fmt.Sprintf("Bad length of trigger: %d (probably normal app)", len(execution.Workflow.Triggers))) } + if execution.Status == "EXECUTING" { + log.Printf("[INFO] Inside executing.") + extra := 0 + for _, trigger := range execution.Workflow.Triggers { + //log.Printf("Appname trigger (0): %s", trigger.AppName) + if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" { + extra += 1 + } + } + + if len(execution.Workflow.Actions)+extra == len(execution.Results) { + execution.Status = "FINISHED" + } + } + // FIXME: Add extra here //executionLength := len(baseExecution.Workflow.Actions) //if executionLength != len(execution.Results) { @@ -986,7 +1004,7 @@ func validateNewWorkerExecution(body []byte) error { //log.Printf("\n\nSHOULD SET BACKEND DATA FOR EXEC \n\n") err = setWorkflowExecution(ctx, execution, true) if err == nil { - log.Printf("[INFO] Set workflowexecution based on new worker (>0.8.53) for execution %s", baseExecution.ExecutionId) + log.Printf("[INFO] Set workflowexecution based on new worker (>0.8.53) for execution %s. Actions: %d, Triggers: %d, Results: %d", execution.ExecutionId, len(execution.Workflow.Actions), len(execution.Workflow.Triggers), len(execution.Results)) //log.Printf("[INFO] Successfully set the execution to wait.") } else { log.Printf("[WARNING] Failed to set the execution to wait.") @@ -3106,7 +3124,7 @@ func cleanupExecutions(resp http.ResponseWriter, request *http.Request) { user, err := handleApiAuthentication(resp, request) if err != nil { - log.Printf("Api authentication failed in execute workflow: %s", err) + log.Printf("[INFO] Api authentication failed in cleanup executions: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false, "message": "Not authenticated"}`)) return @@ -3871,7 +3889,7 @@ func executeWorkflow(resp http.ResponseWriter, request *http.Request) { user, err := handleApiAuthentication(resp, request) if err != nil { - log.Printf("Api authentication failed in execute workflow: %s", err) + log.Printf("[INFO] Api authentication failed in execute workflow: %s", err) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return @@ -6549,8 +6567,20 @@ func iterateAppGithubFolders(fs billy.Filesystem, dir []os.FileInfo, extra strin buildLaterFirst := []buildLaterStruct{} buildLaterList := []buildLaterStruct{} - // It's here to prevent getting them in every iteration ctx := context.Background() + if forceUpdate { + dockercli, err := client.NewEnvClient() + if err == nil { + _, err := dockercli.ImagePull(ctx, "frikky/shuffle:app_sdk", types.ImagePullOptions{}) + if err != nil { + log.Printf("[WARNING] Failed to download apps with the new App SDK: %s", err) + } + } else { + log.Printf("[WARNING] Failed to download apps with the new App SDK because of docker cli: %s", err) + } + } + + // It's here to prevent getting them in every iteration for _, file := range dir { if len(onlyname) > 0 && file.Name() != onlyname { continue diff --git a/docker-compose.yml b/docker-compose.yml index d50df814..ff2bf5d4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,7 +2,7 @@ version: '3' services: frontend: #build: ./frontend - image: ghcr.io/frikky/shuffle-frontend:0.8.61 + image: ghcr.io/frikky/shuffle-frontend:0.8.62 container_name: shuffle-frontend hostname: shuffle-frontend ports: @@ -17,7 +17,7 @@ services: - backend backend: #build: ./backend - image: ghcr.io/frikky/shuffle-backend:0.8.61 + image: ghcr.io/frikky/shuffle-backend:0.8.62 container_name: shuffle-backend hostname: ${BACKEND_HOSTNAME} # Here for debugging: @@ -47,7 +47,7 @@ services: - database orborus: #build: ./functions/onprem/orborus - image: ghcr.io/frikky/shuffle-orborus:0.8.61 + image: ghcr.io/frikky/shuffle-orborus:0.8.62 container_name: shuffle-orborus hostname: shuffle-orborus networks: @@ -56,7 +56,7 @@ services: - /var/run/docker.sock:/var/run/docker.sock environment: - SHUFFLE_APP_SDK_VERSION=0.8.60 - - SHUFFLE_WORKER_VERSION=0.8.61 + - SHUFFLE_WORKER_VERSION=0.8.62 - ORG_ID=${ORG_ID} - ENVIRONMENT_NAME=${ENVIRONMENT_NAME} - BASE_URL=http://${OUTER_HOSTNAME}:${BACKEND_PORT} diff --git a/frontend/src/views/Admin.jsx b/frontend/src/views/Admin.jsx index 8b1d3ec5..7cee4bbe 100644 --- a/frontend/src/views/Admin.jsx +++ b/frontend/src/views/Admin.jsx @@ -377,6 +377,11 @@ const Admin = (props) => { } const handleGetOrg = (orgId) => { + if (orgId.length === 0) { + alert.error("Organization ID not defined. Please contact us on https://shuffler.io if this persists logout.") + return + } + // Just use this one? var baseurl = globalUrl const url = baseurl + '/api/v1/orgs/'+orgId @@ -870,7 +875,8 @@ const Admin = (props) => { }) .then((response) => { if (response.status !== 200) { - window.location.pathname = "/workflows" + // Ahh, this happens because they're not admin + // window.location.pathname = "/workflows" return } @@ -1359,14 +1365,14 @@ const Admin = (props) => { const cancelSubscriptions = (subscription_id) => { console.log(selectedOrganization) + const orgId = selectedOrganization.id const data = { "subscription_id": subscription_id, "action": "cancel", "org_id": selectedOrganization.id, } - - const url = globalUrl + `/api/v1/orgs/${selectedOrganization.id}`; + const url = globalUrl + `/api/v1/orgs/${orgId}`; fetch(url, { mode: 'cors', method: 'POST', diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index f328fbb7..3485f90c 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -382,6 +382,12 @@ const AngularWorkflow = (props) => { if (execution !== null && execution !== undefined) { setExecutionData(execution) setExecutionModalView(1) + start() + + setExecutionRequest({ + "execution_id": execution.execution_id, + "authorization": execution.authorization, + }) const newitem = removeParam("execution_id", cursearch) props.history.push(curpath+newitem) diff --git a/functions/onprem/orborus/build.sh b/functions/onprem/orborus/build.sh index 3156eee0..4c0b9450 100644 --- a/functions/onprem/orborus/build.sh +++ b/functions/onprem/orborus/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-orborus -VERSION=0.8.61 +VERSION=0.8.62 echo "Running docker build with $NAME:$VERSION" #docker rmi frikky/shuffle:$NAME --force diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 4323426b..faae0c38 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -259,7 +259,7 @@ func initializeImages() { log.Printf("[WARNING] SHUFFLE_APP_SDK_VERSION not defined. Defaulting to %s", appSdkVersion) } if workerVersion == "" { - workerVersion = "0.8.61" + workerVersion = "0.8.62" log.Printf("[WARNING] SHUFFLE_WORKER_VERSION not defined. Defaulting to %s", workerVersion) } diff --git a/functions/onprem/worker/Dockerfile b/functions/onprem/worker/Dockerfile index a95b099e..3bf1541a 100644 --- a/functions/onprem/worker/Dockerfile +++ b/functions/onprem/worker/Dockerfile @@ -5,11 +5,11 @@ WORKDIR /app COPY worker.go /app/worker.go RUN go env -w GO111MODULE=auto -RUN go get -u github.com/docker/docker/api/types -RUN go get -u github.com/docker/docker/api/types/container -RUN go get -u github.com/docker/docker/client -RUN go get -u github.com/gorilla/mux -RUN go get -u github.com/patrickmn/go-cache +RUN go get github.com/docker/docker/api/types +RUN go get github.com/docker/docker/api/types/container +RUN go get github.com/docker/docker/client +RUN go get github.com/gorilla/mux +RUN go get github.com/patrickmn/go-cache RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker . diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 10eb7b03..3f2a483e 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-worker -VERSION=0.8.61 +VERSION=0.8.62 echo "Running docker build with $NAME:$VERSION" #CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin . diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 02718009..3e15c282 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -777,11 +777,11 @@ type AppExecutionExample struct { // removes every container except itself (worker) func shutdown(workflowExecution WorkflowExecution, nodeId string, reason string, handleResultSend bool) { - log.Printf("[INFO] Shutdown started with reason %s", reason) + log.Printf("[INFO] Shutdown (%s) started with reason %s", workflowExecution.Status, reason) //reason := "Error in execution" sleepDuration := 1 - if handleResultSend { + if handleResultSend && requestsSent < 2 { data, err := json.Marshal(workflowExecution) if err == nil { sendResult(workflowExecution, data) @@ -829,7 +829,7 @@ func shutdown(workflowExecution WorkflowExecution, nodeId string, reason string, //fmt.Println(url.QueryEscape(query)) fullUrl += path - log.Printf("Abort URL: %s", fullUrl) + log.Printf("[INFO] Abort URL: %s", fullUrl) req, err := http.NewRequest( "GET", @@ -976,7 +976,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] //log.Printf("[INFO] Info for container: %#v", stats) //log.Printf("%#v", stats.Config) //log.Printf("%#v", stats.ContainerJSONBase.State) - log.Printf("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{ ShowStdout: true, @@ -991,6 +991,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] io.Copy(buf, out) logs := buf.String() log.Printf("Logs: %s", logs) + //log.Printf(logs) // check errors /* @@ -2290,6 +2291,8 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`))) return } + + log.Printf(`[INFO] Got result %s from %s`, actionResult.Status, actionResult.Action.ID) resultLength := len(workflowExecution.Results) dbSave := false setExecution := true @@ -2448,6 +2451,7 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl for index, item := range workflowExecution.Results { if item.Action.ID == actionResult.Action.ID { found = true + if item.Status == actionResult.Status { skip = true } @@ -2478,38 +2482,45 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl log.Printf("[INFO] Updating %s in workflow %s from %s to %s", actionResult.Action.ID, workflowExecution.ExecutionId, workflowExecution.Results[outerindex].Status, actionResult.Status) workflowExecution.Results[outerindex] = actionResult } else { - log.Printf("[INFO] Setting value of %s in workflow %s to %s", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status) workflowExecution.Results = append(workflowExecution.Results, actionResult) + log.Printf("[INFO] Setting value (1) of %s in execution %s to %s. New result length: %d", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status, len(workflowExecution.Results)) } } else { - log.Printf("[INFO] Setting value of %s in workflow %s to %s", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status) workflowExecution.Results = append(workflowExecution.Results, actionResult) + log.Printf("[INFO] Setting value (2) of %s in execution %s to %s. New result length: %d", actionResult.Action.ID, workflowExecution.ExecutionId, actionResult.Status, len(workflowExecution.Results)) } // FIXME: Have a check for skippednodes and their parents - for resultIndex, result := range workflowExecution.Results { - if result.Status != "SKIPPED" { - continue - } + /* + for resultIndex, result := range workflowExecution.Results { + if result.Status != "SKIPPED" { + continue + } - // Checks if all parents are skipped or failed. Otherwise removes them from the results - for _, branch := range workflowExecution.Workflow.Branches { - if branch.DestinationID == result.Action.ID { - for _, subresult := range workflowExecution.Results { - if subresult.Action.ID == branch.SourceID { - if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" { - log.Printf("SUBRESULT PARENT STATUS: %s", subresult.Status) - log.Printf("Should remove resultIndex: %d", resultIndex) + // Checks if all parents are skipped or failed. + // Otherwise removes them from the results + for _, branch := range workflowExecution.Workflow.Branches { + if branch.DestinationID == result.Action.ID { + for _, subresult := range workflowExecution.Results { + if subresult.Action.ID == branch.SourceID { + if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" { + //log.Printf("SUBRESULT PARENT STATUS: %s", subresult.Status) + //log.Printf("Should remove resultIndex: %d", resultIndex) - workflowExecution.Results = append(workflowExecution.Results[:resultIndex], workflowExecution.Results[resultIndex+1:]...) + // FIXME: Reinstate this? + //workflowExecution.Results = append(workflowExecution.Results[:resultIndex], workflowExecution.Results[resultIndex+1:]...) + _ = resultIndex - break + break + } } } } } } - } + + log.Printf("NEW LENGTH: %d", len(workflowExecution.Results)) + */ extraInputs := 0 for _, trigger := range workflowExecution.Workflow.Triggers { @@ -2697,7 +2708,7 @@ func sendResult(workflowExecution WorkflowExecution, data []byte) { } func validateFinished(workflowExecution WorkflowExecution) { - log.Printf("[INFO] Status: %s, Actions: %d, Extra: %d, Results: %d\n", workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results)) + log.Printf("[INFO] VALIDATION. Status: %s, Actions: %d, Extra: %d, Results: %d\n", workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results)) //if len(workflowExecution.Results) == len(workflowExecution.Workflow.Actions)+extra { if (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1) || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions) && len(workflowExecution.Workflow.Actions) > 0) {