Optimization: Made apps and executions way faster

backend/app_sdk/Dockerfile

@@ -7,7 +7,7 @@ RUN mkdir /install
 WORKDIR /install
 
 COPY requirements.txt /requirements.txt
-RUN pip install --prefix="/install" -r /requirements.txt
+RUN pip3 install -r /requirements.txt
 
 FROM base
 

backend/app_sdk/app_base.py

@@ -21,6 +21,7 @@ class AppBase:
         # apikey is for the user / org
         # authorization is for the specific workflow
         self.url = os.getenv("CALLBACK_URL", "https://shuffler.io")
+        self.base_url = os.getenv("BASE_URL", "https://shuffler.io")
         self.action = os.getenv("ACTION", "")
         self.authorization = os.getenv("AUTHORIZATION", "")
         self.current_execution_id = os.getenv("EXECUTIONID", "")
@@ -30,6 +31,9 @@ class AppBase:
         if isinstance(self.action, str):
             self.action = json.loads(self.action)
 
+        if len(self.base_url) == 0:
+            self.base_url = self.url
+
     # FIXME: Add more info like logs in here.
     # Docker logs: https://forums.docker.com/t/docker-logs-inside-the-docker-container/68190/2
     def send_result(self, action_result, headers, stream_path):
@@ -38,7 +42,7 @@ class AppBase:
 
         # I wonder if this actually works 
         self.logger.info("Before last stream result")
-        url = "%s%s" % (self.url, stream_path)
+        url = "%s%s" % (self.base_url, stream_path)
         print("URL: %s" % url)
         try:
             ret = requests.post(url, headers=headers, json=action_result)
@@ -53,7 +57,7 @@ class AppBase:
             action_result["status"] = "FAILURE"
             action_result["result"] = "POST error: %s" % e
             self.logger.info("Before typeerror stream result")
-            ret = requests.post("%s%s" % (self.url, stream_path), headers=headers, json=action_result)
+            ret = requests.post("%s%s" % (self.base_url, stream_path), headers=headers, json=action_result)
             self.logger.info("Result: %d" % ret.status_code)
             if ret.status_code != 200:
                 self.logger.info(ret.text)
@@ -536,7 +540,7 @@ class AppBase:
 
         # FIXME: Shouldn't skip this, but it's good for minimzing API calls
         #try:
-        #    ret = requests.post("%s%s" % (self.url, stream_path), headers=headers, json=action_result)
+        #    ret = requests.post("%s%s" % (self.base_url, stream_path), headers=headers, json=action_result)
         #    self.logger.info("Workflow: %d" % ret.status_code)
         #    if ret.status_code != 200:
         #        self.logger.info(ret.text)
@@ -560,7 +564,7 @@ class AppBase:
 
                 self.logger.info("Before FULLEXEC stream result")
                 ret = requests.post(
-                    "%s/api/v1/streams/results" % (self.url), 
+                    "%s/api/v1/streams/results" % (self.base_url), 
                     headers=headers, 
                     json=tmpdata
                 )
@@ -568,8 +572,12 @@ class AppBase:
                 if ret.status_code == 200:
                     fullexecution = ret.json()
                 else:
-                    self.logger.info("Error: Data: ", ret.json())
+                    try:
-                    self.logger.info("Error with status code for results. Crashing because ACTION_RESULTS or WORKFLOW_VARIABLE can't be handled. Status: %d" % ret.status_code)
+                        self.logger.info("Error: Data: ", ret.json())
+                        self.logger.info("Error with status code for results. Crashing because ACTION_RESULTS or WORKFLOW_VARIABLE can't be handled. Status: %d" % ret.status_code)
+                    except json.decoder.JSONDecodeError:
+                        pass
+
                     action_result["result"] = "Bad result from backend: %d" % ret.status_code
                     self.send_result(action_result, headers, stream_path) 
                     return
@@ -1322,7 +1330,7 @@ class AppBase:
             action_result["result"] = tmpresult
             action_result["status"] = "FAILURE"
             try:
-                ret = requests.post("%s%s" % (self.url, stream_path), headers=headers, json=action_result)
+                ret = requests.post("%s%s" % (self.base_url, stream_path), headers=headers, json=action_result)
                 self.logger.info("Result: %d" % ret.status_code)
                 if ret.status_code != 200:
                     self.logger.info(ret.text)

backend/app_sdk/build.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 NAME=shuffle-app_sdk
-VERSION=0.8.51
+VERSION=0.8.52
 
 docker rmi docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION --force
 docker build . -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION
@@ -8,6 +8,7 @@ docker build . -t frikky/shuffle:app_sdk -t frikky/$NAME:$VERSION -t docker.pkg.
 #docker push frikky/$NAME:$VERSION
 #docker push docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION
 #docker push ghcr.io/frikky/$NAME:$VERSION
+#docker tag ghcr.io/frikky/$NAME:$VERSION frikky/shuffle:app_sdk
 
 docker push frikky/shuffle:app_sdk
 docker push ghcr.io/frikky/$NAME:$VERSION

backend/app_sdk/requirements.txt

@@ -1,2 +1,2 @@
-requests
 urllib3
+requests

backend/go-app/docker.go

@@ -246,6 +246,7 @@ func buildImageMemory(fs billy.Filesystem, tags []string, dockerfileFolder strin
 	if newerr != nil {
 		log.Printf("Failed reading Docker build STDOUT: %s", newerr)
 	} else {
+		log.Printf("STRING: %s", buildBuf.String())
 		if strings.Contains(buildBuf.String(), "errorDetail") {
 			log.Printf("[ERROR] Docker build:\n%s\nERROR ABOVE: Trying to pull tags from: %s", buildBuf.String(), strings.Join(tags, "\n"))
 

backend/go-app/files.go

@@ -43,6 +43,8 @@ type File struct {
 	Md5sum       string   `json:"md5_sum" datastore:"md5_sum"`
 	Sha256sum    string   `json:"sha256_sum" datastore:"sha256_sum"`
 	FileSize     int64    `json:"filesize" datastore:"filesize"`
+	Duplicate    bool     `json:"duplicate" datastore:"duplicate"`
+	Subflows     []string `json:"subflows" datastore:"subflows"`
 }
 
 var basepath = os.Getenv("SHUFFLE_FILE_LOCATION")
@@ -393,7 +395,7 @@ func handleGetFileContent(resp http.ResponseWriter, request *http.Request) {
 		return
 	}
 
-	log.Printf("\n\nUser is trying to download file %s\n\n", fileId)
+	log.Printf("\n\n[INFO] User is trying to download file %s\n\n", fileId)
 
 	// 1. Check user directly
 	// 2. Check workflow execution authorization
@@ -772,6 +774,30 @@ func handleCreateFile(resp http.ResponseWriter, request *http.Request) {
 	fileId := uuid.NewV4().String()
 	downloadPath := fmt.Sprintf("%s/%s", folderPath, fileId)
 
+	duplicateWorkflows := []string{}
+	for _, trigger := range workflow.Triggers {
+		if trigger.AppName == "Shuffle Workflow" && trigger.TriggerType == "SUBFLOW" {
+			for _, parameter := range trigger.Parameters {
+				if parameter.Name == "workflow" && len(parameter.Value) > 0 {
+
+					found := false
+					for _, workflow := range duplicateWorkflows {
+						if workflow == parameter.Value {
+							found = true
+							break
+						}
+					}
+
+					if !found {
+						duplicateWorkflows = append(duplicateWorkflows, parameter.Value)
+					}
+
+					break
+				}
+			}
+		}
+	}
+
 	timeNow := time.Now().Unix()
 	newFile := File{
 		Id:           fileId,
@@ -783,6 +809,7 @@ func handleCreateFile(resp http.ResponseWriter, request *http.Request) {
 		OrgId:        curfile.OrgId,
 		WorkflowId:   curfile.WorkflowId,
 		DownloadPath: downloadPath,
+		Subflows:     duplicateWorkflows,
 	}
 
 	err = setFile(ctx, newFile)
@@ -797,6 +824,7 @@ func handleCreateFile(resp http.ResponseWriter, request *http.Request) {
 
 	resp.WriteHeader(200)
 	resp.Write([]byte(fmt.Sprintf(`{"success": true, "id": "%s"}`, fileId)))
+
 }
 
 func getFile(ctx context.Context, id string) (*File, error) {

backend/go-app/walkoff.go

@@ -893,6 +893,53 @@ func findChildNodes(workflowExecution WorkflowExecution, nodeId string) []string
 	return newNodes
 }
 
+// Checks if data is sent from Worker >0.8.51, which sends a full execution
+// instead of individial results
+func validateNewWorkerExecution(body []byte) error {
+	//type WorkflowExecution struct {
+	//}
+
+	ctx := context.Background()
+	var execution WorkflowExecution
+	err := json.Unmarshal(body, &execution)
+	if err != nil {
+		log.Printf("[WARNING] Failed execution unmarshaling: %s", err)
+		return err
+	}
+
+	baseExecution, err := getWorkflowExecution(ctx, execution.ExecutionId)
+	if err != nil {
+		log.Printf("[ERROR] Failed getting execution (workflowqueue) %s: %s", execution.ExecutionId, err)
+		return err
+	}
+
+	if baseExecution.Authorization != execution.Authorization {
+		return errors.New("Bad authorization when validating execution")
+	}
+
+	if len(baseExecution.Workflow.Actions) != len(execution.Workflow.Actions) {
+		return errors.New(fmt.Sprintf("Bad length of actions: %d", len(execution.Workflow.Actions)))
+	}
+
+	if len(baseExecution.Workflow.Triggers) != len(execution.Workflow.Triggers) {
+		return errors.New(fmt.Sprintf("Bad length of trigger: %d", len(execution.Workflow.Triggers)))
+	}
+
+	// FIXME: Add extra here
+	//executionLength := len(baseExecution.Workflow.Actions)
+	//if executionLength != len(execution.Results) {
+	//	return errors.New(fmt.Sprintf("Bad length of actions vs results: want: %d have: %d", executionLength, len(execution.Results)))
+	//}
+
+	//log.Printf("\n\nSHOULD SET BACKEND DATA FOR EXEC \n\n")
+	err = setWorkflowExecution(ctx, execution, true)
+	if err != nil {
+		log.Printf("Successfully set the execution to wait.")
+	}
+
+	return nil
+}
+
 func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 	cors := handleCors(resp, request)
 	if cors {
@@ -907,6 +954,17 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 		return
 	}
 
+	//log.Printf("Actionresult unmarshal: %s", string(body))
+	err = validateNewWorkerExecution(body)
+	if err == nil {
+		log.Printf("[INFO] Set workflowexecution based on new worker")
+		resp.WriteHeader(200)
+		resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Success"}`)))
+		return
+	} else {
+		log.Printf("[WARNING] Failed to handle new execution variant: %s", err)
+	}
+
 	var actionResult ActionResult
 	err = json.Unmarshal(body, &actionResult)
 	if err != nil {
@@ -984,7 +1042,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
 			workflowExecution.Status = "ABORTED"
 			err = setWorkflowExecution(ctx, *workflowExecution, true)
 			if err != nil {
-				log.Printf("Failed ")
+				log.Printf("Failed to set execution during wait")
 			} else {
 				log.Printf("Successfully set the execution to waiting.")
 			}
@@ -2985,7 +3043,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 
 	allAuths := []AppAuthenticationStorage{}
 	for _, action := range workflowExecution.Workflow.Actions {
-		action.LargeImage = ""
+		//action.LargeImage = ""
 		if action.ID == workflowExecution.Start {
 			startFound = true
 		}
@@ -3036,6 +3094,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
 			action.Parameters = newParams
 		}
 
+		action.LargeImage = ""
 		newActions = append(newActions, action)
 
 		// If the node is NOT found, it's supposed to be set to SKIPPED,

frontend/src/views/AngularWorkflow.jsx

@@ -615,9 +615,11 @@ const AngularWorkflow = (props) => {
 
 			getWorkflowExecution(props.match.params.key)
 		} else if (responseJson.status === "FINISHED") {
+			console.log("STOPPING BECAUSE ITS OVAH!")
 			setExecutionRunning(false)
 			stop()
 			getWorkflowExecution(props.match.params.key)
+			setUpdate(Math.random())
 		}
 	}
 

functions/onprem/orborus/orborus.go

@@ -248,7 +248,7 @@ func initializeImages() {
 		log.Printf("[WARNING] SHUFFLE_APP_SDK_VERSION not defined. Defaulting to %s", appSdkVersion)
 	}
 	if workerVersion == "" {
-		workerVersion = "0.8.5"
+		workerVersion = "0.8.52"
 		log.Printf("[WARNING] SHUFFLE_WORKER_VERSION not defined. Defaulting to %s", workerVersion)
 	}
 

functions/onprem/worker/build.sh

@@ -1,5 +1,5 @@
 NAME=shuffle-worker
-VERSION=0.8.51
+VERSION=0.8.52
 
 echo "Running docker build with $NAME:$VERSION"
 #CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin .
@@ -9,5 +9,6 @@ docker build . -t frikky/shuffle:$NAME -t frikky/shuffle:$NAME_$VERSION -t docke
 #docker push frikky/$NAME:$VERSION 
 #docker push frikky/shuffle:$NAME_$VERSION
 #docker push docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION
-docker tag frikky/shuffle:0.8.51 ghcr.io/frikky/shuffle-worker:0.8.5
+#docker tag frikky/shuffle:0.8.51 ghcr.io/frikky/shuffle-worker:0.8.5
-docker push ghcr.io/frikky/$NAME:$VERSION
+#docker push ghcr.io/frikky/$NAME:$VERSION
+docker tag frikky/shuffle:$NAME ghcr.io/frikky/shuffle-worker:0.8.52

functions/onprem/worker/worker.go

@@ -2130,7 +2130,7 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl
 	//}
 
 	if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" {
-		dbSave = true
+		//dbSave = true
 
 		newResults := []ActionResult{}
 		childNodes := []string{}
@@ -2576,6 +2576,9 @@ func setWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecuti
 
 	handleExecutionResult(workflowExecution)
 	validateFinished(workflowExecution)
+	if dbSave {
+		shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID)
+	}
 	return nil
 }