From 34097f971eb8756c4e217090d4227c6aea83b82f Mon Sep 17 00:00:00 2001 From: frikky Date: Tue, 12 Jan 2021 18:29:27 +0100 Subject: [PATCH] Optimization: Made apps and executions way faster --- backend/app_sdk/Dockerfile | 2 +- backend/app_sdk/app_base.py | 22 ++++++--- backend/app_sdk/build.sh | 3 +- backend/app_sdk/requirements.txt | 2 +- backend/go-app/docker.go | 1 + backend/go-app/files.go | 30 +++++++++++- backend/go-app/walkoff.go | 63 +++++++++++++++++++++++++- frontend/src/views/AngularWorkflow.jsx | 2 + functions/onprem/orborus/orborus.go | 2 +- functions/onprem/worker/build.sh | 7 +-- functions/onprem/worker/worker.go | 5 +- 11 files changed, 121 insertions(+), 18 deletions(-) diff --git a/backend/app_sdk/Dockerfile b/backend/app_sdk/Dockerfile index 44294a5b..370c31f8 100644 --- a/backend/app_sdk/Dockerfile +++ b/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 diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index 8560f1ce..d88b7626 100644 --- a/backend/app_sdk/app_base.py +++ b/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()) - 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) + try: + 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) diff --git a/backend/app_sdk/build.sh b/backend/app_sdk/build.sh index 5a24cba5..c6911f6e 100644 --- a/backend/app_sdk/build.sh +++ b/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 diff --git a/backend/app_sdk/requirements.txt b/backend/app_sdk/requirements.txt index 804abb1b..0da82340 100644 --- a/backend/app_sdk/requirements.txt +++ b/backend/app_sdk/requirements.txt @@ -1,2 +1,2 @@ -requests urllib3 +requests diff --git a/backend/go-app/docker.go b/backend/go-app/docker.go index ccacd3ea..f05e4d38 100644 --- a/backend/go-app/docker.go +++ b/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")) diff --git a/backend/go-app/files.go b/backend/go-app/files.go index b9b30a6d..7312af1f 100644 --- a/backend/go-app/files.go +++ b/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) { diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index c42baf98..0b06a1bb 100644 --- a/backend/go-app/walkoff.go +++ b/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, diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 80228f8c..ae6e8302 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/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()) } } diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 231f12a4..42e9d097 100644 --- a/functions/onprem/orborus/orborus.go +++ b/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) } diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 2b0a8a0a..29828da4 100644 --- a/functions/onprem/worker/build.sh +++ b/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 push ghcr.io/frikky/$NAME:$VERSION +#docker tag frikky/shuffle:0.8.51 ghcr.io/frikky/shuffle-worker:0.8.5 +#docker push ghcr.io/frikky/$NAME:$VERSION +docker tag frikky/shuffle:$NAME ghcr.io/frikky/shuffle-worker:0.8.52 diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index d77eb362..49c4b583 100644 --- a/functions/onprem/worker/worker.go +++ b/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 }