Workflow completion fixes

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

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}

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/${orgId}`;
-		const url = globalUrl + `/api/v1/orgs/${selectedOrganization.id}`;
 		fetch(url, {
 			mode: 'cors',
 			method: 'POST',

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)

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

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)
 	}
 

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 github.com/docker/docker/api/types 
-RUN go get -u github.com/docker/docker/api/types/container
+RUN go get github.com/docker/docker/api/types/container
-RUN go get -u github.com/docker/docker/client
+RUN go get github.com/docker/docker/client
-RUN go get -u github.com/gorilla/mux
+RUN go get github.com/gorilla/mux
-RUN go get -u github.com/patrickmn/go-cache
+RUN go get github.com/patrickmn/go-cache
 
 RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker .
 

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 .

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" {
+		for resultIndex, result := range workflowExecution.Results {
-			continue
+			if result.Status != "SKIPPED" {
-		}
+				continue
+			}
 
-		// Checks if all parents are skipped or failed. Otherwise removes them from the results
+			// Checks if all parents are skipped or failed.
-		for _, branch := range workflowExecution.Workflow.Branches {
+			// Otherwise removes them from the results
-			if branch.DestinationID == result.Action.ID {
+			for _, branch := range workflowExecution.Workflow.Branches {
-				for _, subresult := range workflowExecution.Results {
+				if branch.DestinationID == result.Action.ID {
-					if subresult.Action.ID == branch.SourceID {
+					for _, subresult := range workflowExecution.Results {
-						if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" {
+						if subresult.Action.ID == branch.SourceID {
-							log.Printf("SUBRESULT PARENT STATUS: %s", subresult.Status)
+							if subresult.Status != "SKIPPED" && subresult.Status != "FAILURE" {
-							log.Printf("Should remove resultIndex: %d", resultIndex)
+								//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) {