diff --git a/backend/go-app/docker.go b/backend/go-app/docker.go index 06b9b086..82ff6f5f 100644 --- a/backend/go-app/docker.go +++ b/backend/go-app/docker.go @@ -721,7 +721,7 @@ func handleStartHookDocker(resp http.ResponseWriter, request *http.Request) { // Checks if an image exists func imageCheckBuilder(images []string) error { - log.Printf("[FIXME] ImageNames to check: %#v", images) + //log.Printf("[FIXME] ImageNames to check: %#v", images) return nil ctx := context.Background() diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index 553b8b3c..b9bbe755 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -23,6 +23,7 @@ require ( github.com/gorilla/mux v1.7.4 github.com/h2non/filetype v1.0.12 github.com/opencontainers/go-digest v1.0.0-rc1 // indirect + github.com/patrickmn/go-cache v2.1.0+incompatible // indirect github.com/satori/go.uuid v1.2.0 golang.org/x/crypto v0.0.0-20200429183012-4b2356b1ed79 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d diff --git a/backend/go-app/go.sum b/backend/go-app/go.sum index 4fbd4b69..2d63be91 100644 --- a/backend/go-app/go.sum +++ b/backend/go-app/go.sum @@ -160,6 +160,8 @@ github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrk github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= github.com/opencontainers/go-digest v1.0.0-rc1 h1:WzifXhOVOEOuFYOJAW6aQqW0TooG2iki3E3Ii+WN7gQ= github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s= +github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc= +github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ= github.com/pelletier/go-buffruneio v0.2.0/go.mod h1:JkE26KsDizTr40EUHkXVtNPvgGtbSNq5BcowyYOWdKo= github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= diff --git a/backend/go-app/main.go b/backend/go-app/main.go index d6ebc6cb..8fe7b20e 100644 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -65,6 +65,7 @@ import ( // "google.golang.org/appengine/memcache" // applog "google.golang.org/appengine/log" //cloudrun "google.golang.org/api/run/v1" + "github.com/patrickmn/go-cache" ) // This is used to handle onprem vs offprem databases etc @@ -80,6 +81,7 @@ var syncUrl = "https://shuffler.io" //var syncUrl = "http://localhost:5002" var dbclient *datastore.Client +var requestCache *cache.Cache type Userapi struct { Username string `datastore:"username"` @@ -2634,6 +2636,23 @@ func getUser(ctx context.Context, id string) (*User, error) { return curUser, nil } +// Index = Username +func DeleteKeys(ctx context.Context, entity string, value []string) error { + // Non indexed User data + keys := []*datastore.Key{} + for _, item := range value { + keys = append(keys, datastore.NameKey(entity, item, nil)) + } + + err := dbclient.DeleteMulti(ctx, keys) + if err != nil { + log.Printf("Error deleting %s from %s: %s", value, entity, err) + return err + } + + return nil +} + // Index = Username func DeleteKey(ctx context.Context, entity string, value string) error { // Non indexed User data @@ -6776,7 +6795,7 @@ func handleCloudJob(job CloudSyncJob) error { } workflowExecution.Status = "EXECUTING" - err = setWorkflowExecution(ctx, *workflowExecution) + err = setWorkflowExecution(ctx, *workflowExecution, true) if err != nil { return err } @@ -6829,7 +6848,7 @@ func handleCloudJob(job CloudSyncJob) error { workflowExecution.Results = newResults workflowExecution.Status = "ABORTED" - err = setWorkflowExecution(ctx, *workflowExecution) + err = setWorkflowExecution(ctx, *workflowExecution, true) if err != nil { return err } @@ -6960,6 +6979,8 @@ func runInit(ctx context.Context) { log.Printf("Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy) } + requestCache = cache.New(5*time.Minute, 10*time.Minute) + /* proxyUrl, err := url.Parse(httpProxy) if err != nil { diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index adb69f94..560a2178 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -36,6 +36,8 @@ import ( //"google.golang.org/appengine/memcache" //"cloud.google.com/go/firestore" // "google.golang.org/api/option" + + "github.com/patrickmn/go-cache" ) var localBase = "http://localhost:5001" @@ -494,11 +496,12 @@ func increaseStatisticsField(ctx context.Context, fieldname, id string, amount i return nil } -func setWorkflowQueue(ctx context.Context, executionRequests ExecutionRequestWrapper, id string) error { - key := datastore.NameKey("workflowqueue", id, nil) +func setWorkflowQueue(ctx context.Context, executionRequest ExecutionRequest, env string) error { + orgKey := fmt.Sprintf("workflowqueue-%s", env) + key := datastore.NameKey(orgKey, executionRequest.ExecutionId, nil) // New struct, to not add body, author etc - if _, err := dbclient.Put(ctx, key, &executionRequests); err != nil { + if _, err := dbclient.Put(ctx, key, &executionRequest); err != nil { log.Printf("Error adding workflow queue: %s", err) return err } @@ -506,14 +509,37 @@ func setWorkflowQueue(ctx context.Context, executionRequests ExecutionRequestWra return nil } +// +//func setWorkflowQueue(ctx context.Context, executionRequests ExecutionRequestWrapper, id string) error { +// key := datastore.NameKey("workflowqueue", id, nil) +// +// // New struct, to not add body, author etc +// if _, err := dbclient.Put(ctx, key, &executionRequests); err != nil { +// log.Printf("Error adding workflow queue: %s", err) +// return err +// } +// +// return nil +//} + func getWorkflowQueue(ctx context.Context, id string) (ExecutionRequestWrapper, error) { - key := datastore.NameKey("workflowqueue", id, nil).Limit(50) - workflows := ExecutionRequestWrapper{} - if err := dbclient.Get(ctx, key, &workflows); err != nil { + orgId := fmt.Sprintf("workflowqueue-%s", id) + q := datastore.NewQuery(orgId).Limit(10) + executions := []ExecutionRequest{} + _, err := dbclient.GetAll(ctx, q, &executions) + if err != nil { return ExecutionRequestWrapper{}, err } - return workflows, nil + return ExecutionRequestWrapper{Data: executions}, nil + + //key := datastore.NameKey("workflowqueue", id, nil) + //executions := ExecutionRequestWrapper{} + //if err := dbclient.Get(ctx, key, &workflows); err != nil { + // return ExecutionRequestWrapper{}, err + //} + + //return workflows, nil } //func setWorkflowqueuetest(id string) { @@ -649,9 +675,9 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque } if len(executionRequests.Data) == 0 { - log.Printf("No requests to fix. Why did this request occur?") - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Some error"}`))) + log.Printf("[INFO] No requests to handle from queue") + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Nothing in queue"}`))) return } @@ -675,41 +701,47 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque } if len(removeExecutionRequests.Data) == 0 { - log.Printf("No requests to fix remove") + log.Printf("No requests to fix remove from DB") resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Some removal error"}`))) return } // remove items from DB - var newExecutionRequests ExecutionRequestWrapper - for _, execution := range executionRequests.Data { - found := false - for _, removeExecution := range removeExecutionRequests.Data { - if removeExecution.ExecutionId == execution.ExecutionId && removeExecution.WorkflowId == execution.WorkflowId { - found = true - break - } - } - - if !found { - newExecutionRequests.Data = append(newExecutionRequests.Data, execution) - } + parsedId := fmt.Sprintf("workflowqueue-%s", id) + ids := []string{} + for _, execution := range removeExecutionRequests.Data { + ids = append(ids, execution.ExecutionId) } + err = DeleteKeys(ctx, parsedId, ids) + if err != nil { + log.Printf("[ERROR] Failed deleting %d execution keys for org %s", len(ids), id) + } else { + //log.Printf("[INFO] Deleted %d keys from org %s", len(ids), parsedId) + } + + //var newExecutionRequests ExecutionRequestWrapper + //for _, execution := range executionRequests.Data { + // found := false + // for _, removeExecution := range removeExecutionRequests.Data { + // if removeExecution.ExecutionId == execution.ExecutionId && removeExecution.WorkflowId == execution.WorkflowId { + // found = true + // break + // } + // } + + // if !found { + // newExecutionRequests.Data = append(newExecutionRequests.Data, execution) + // } + //} + // Push only the remaining to the DB (remove) - if len(executionRequests.Data) != len(newExecutionRequests.Data) { - err := setWorkflowQueue(ctx, newExecutionRequests, id) - if err != nil { - log.Printf("Fail: %s", err) - } - } - - //newjson, err := json.Marshal(removeExecutionRequests) - //if err != nil { - // resp.WriteHeader(401) - // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed unpacking workflow execution"}`))) - // return + //if len(executionRequests.Data) != len(newExecutionRequests.Data) { + // err := setWorkflowQueue(ctx, newExecutionRequests, id) + // if err != nil { + // log.Printf("Fail: %s", err) + // } //} resp.WriteHeader(200) @@ -892,14 +924,14 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { ctx := context.Background() workflowExecution, err := getWorkflowExecution(ctx, actionResult.ExecutionId) if err != nil { - log.Printf("Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, err) + log.Printf("[ERROR] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, err) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist."}`, actionResult.ExecutionId))) return } if workflowExecution.Authorization != actionResult.Authorization { - log.Printf("Bad authorization key when updating node (workflowQueue) %s. Want: %s, Have: %s", actionResult.ExecutionId, workflowExecution.Authorization, actionResult.Authorization) + log.Printf("[INFO] Bad authorization key when updating node (workflowQueue) %s. Want: %s, Have: %s", actionResult.ExecutionId, workflowExecution.Authorization, actionResult.Authorization) resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key"}`))) return @@ -950,7 +982,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { actionResult.Result = fmt.Sprintf("Cloud error: %s", err) workflowExecution.Results = append(workflowExecution.Results, actionResult) workflowExecution.Status = "ABORTED" - err = setWorkflowExecution(ctx, *workflowExecution) + err = setWorkflowExecution(ctx, *workflowExecution, true) if err != nil { log.Printf("Failed ") } else { @@ -968,7 +1000,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { workflowExecution.Results = append(workflowExecution.Results, actionResult) workflowExecution.Status = actionResult.Status - err = setWorkflowExecution(ctx, *workflowExecution) + err = setWorkflowExecution(ctx, *workflowExecution, true) if err != nil { log.Printf("Failed ") } else { @@ -985,26 +1017,34 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { // Will make sure transactions are always ran for an execution. This is recursive if it fails. Allowed to fail up to 5 times func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workflowExecutionId string, actionResult ActionResult, resp http.ResponseWriter) { // Should start a tx for the execution here - tx, err := dbclient.NewTransaction(ctx) + workflowExecution, err := getWorkflowExecution(ctx, workflowExecutionId) if err != nil { - log.Printf("client.NewTransaction: %v", err) + log.Printf("[ERROR] Failed getting execution cache: %s", err) resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed creating transaction"}`))) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`))) return } + resultLength := len(workflowExecution.Results) + //tx, err := dbclient.NewTransaction(ctx) + //if err != nil { + // log.Printf("client.NewTransaction: %v", err) + // resp.WriteHeader(401) + // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed creating transaction"}`))) + // return + //} - key := datastore.NameKey("workflowexecution", workflowExecutionId, nil) - workflowExecution := &WorkflowExecution{} - if err := tx.Get(key, workflowExecution); err != nil { - log.Printf("tx.Get bug: %v", err) - tx.Rollback() - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting the workflow key"}`))) - return - } + //key := datastore.NameKey("workflowexecution", workflowExecutionId, nil) + //workflowExecution := &WorkflowExecution{} + //if err := tx.Get(key, workflowExecution); err != nil { + // log.Printf("[ERROR] tx.Get bug: %v", err) + // tx.Rollback() + // resp.WriteHeader(401) + // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting the workflow key"}`))) + // return + //} if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" { - log.Printf("[WARNING] Actionresult is %s. Should set workflowExecution and exit all running functions", actionResult.Status) + log.Printf("[WARNING] Actionresult is %s for node %s in %s. Should set workflowExecution and exit all running functions", actionResult.Status, actionResult.Action.ID, workflowExecution.ExecutionId) newResults := []ActionResult{} childNodes := []string{} @@ -1277,52 +1317,79 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl } } - // Transactions: https://cloud.google.com/datastore/docs/concepts/transactions#datastore-datastore-transactional-update-go - // Prevents timing issues - //ExecutionId - if _, err := tx.Put(key, workflowExecution); err != nil { - log.Printf("[ERROR] tx.Put error: %v", err) - err = tx.Rollback() - if err != nil { - log.Printf("[ERROR] Rollback error (3): %s", err) - } + // Validating that action results hasn't changed + // Handled using cachhing, so actually pretty fast + setExecution := true + cacheKey := fmt.Sprintf("workflowexecution-%s", workflowExecution.ExecutionId) + if value, found := requestCache.Get(cacheKey); found { + parsedValue := value.(*WorkflowExecution) + if len(parsedValue.Results) > 0 && len(parsedValue.Results) != resultLength { + setExecution = false + if attempts > 5 { + log.Printf("\n\nSkipping execution input - %d vs %d. Attempts: (%d)\n\n", len(parsedValue.Results), resultLength, attempts) + } - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err))) - return - } - - if _, err = tx.Commit(); err != nil { - err = tx.Rollback() - if err != nil { - log.Printf("[ERROR] Rollback error expected ? (1): %s", err) - } - - if attempts >= 7 { - log.Printf("[ERROR] QUITTING: tx.Commit %d: %v", attempts, err) - - workflowExecution.Status = "ABORTED" - setWorkflowExecution(ctx, *workflowExecution) - - resp.WriteHeader(401) - resp.Write([]byte(`{"success": false}`)) + attempts += 1 + runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) return } - - if attempts > 3 { - log.Printf("[WARNING] tx.Commit %d: %v", attempts, err) - } - - attempts += 1 - runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) - return - } else { - //if grpc.Code(err) == codes.Aborted { - // return nil, ErrConcurrentTransaction - //} - //t.id = nil // mark the transaction as expired } + if setExecution { + err = setWorkflowExecution(ctx, *workflowExecution, false) + if err != nil { + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err))) + return + } + } + + //ExecutionId + // Transactions: https://cloud.google.com/datastore/docs/concepts/transactions#datastore-datastore-transactional-update-go + // Prevents timing issues + //if _, err := tx.Put(key, workflowExecution); err != nil { + // log.Printf("[ERROR] tx.Put error: %v", err) + // err = tx.Rollback() + // if err != nil { + // log.Printf("[ERROR] Rollback error (3): %s", err) + // } + + // resp.WriteHeader(401) + // resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed setting workflowexecution actionresult: %s"}`, err))) + // return + //} + + //if _, err = tx.Commit(); err != nil { + // err = tx.Rollback() + // if err != nil { + // log.Printf("[ERROR] Rollback error expected ? (1): %s", err) + // } + + // if attempts >= 7 { + // log.Printf("[ERROR] QUITTING: tx.Commit %d: %v", attempts, err) + + // workflowExecution.Status = "ABORTED" + // setWorkflowExecution(ctx, *workflowExecution, true) + + // resp.WriteHeader(401) + // resp.Write([]byte(`{"success": false}`)) + // return + // } + + // if attempts > 3 { + // log.Printf("[WARNING] tx.Commit %d: %v", attempts, err) + // } + + // attempts += 1 + // runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) + // return + //} else { + // //if grpc.Code(err) == codes.Aborted { + // // return nil, ErrConcurrentTransaction + // //} + // //t.id = nil // mark the transaction as expired + //} + resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": true}`))) } @@ -1407,7 +1474,7 @@ func handleExecutionStatistics(execution WorkflowExecution) { log.Printf("[INFO] Added %d exampleresults to backend", successful) } else { - log.Printf("[INFO] No example results necessary to be added for execution %s", execution.ExecutionId) + //log.Printf("[INFO] No example results necessary to be added for execution %s", execution.ExecutionId) } } @@ -2458,7 +2525,7 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) { return } } else { - log.Printf("[INFO] API key to abort/finish execution %s is correct.", executionId) + //log.Printf("[INFO] API key to abort/finish execution %s is correct.", executionId) } if workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" || workflowExecution.Status == "FINISHED" { @@ -2494,7 +2561,7 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) { workflowExecution.Result = lastResult } - err = setWorkflowExecution(ctx, *workflowExecution) + err = setWorkflowExecution(ctx, *workflowExecution, true) if err != nil { log.Printf("Error saving workflow execution for updates when aborting %s: %s", topic, err) resp.WriteHeader(401) @@ -2614,12 +2681,12 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf if request.Method == "POST" { body, err := ioutil.ReadAll(request.Body) if err != nil { - log.Printf("Failed request POST read: %s", err) + log.Printf("[ERROR] Failed request POST read: %s", err) return WorkflowExecution{}, "Failed getting body", err } // This one doesn't really matter. - log.Printf("Running POST execution with body of length %d", len(string(body))) + log.Printf("[INFO] Running POST execution with body of length %d", len(string(body))) var execution ExecutionRequest err = json.Unmarshal(body, &execution) if err != nil { @@ -2718,7 +2785,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf } oldExecution.Results = newResults - err = setWorkflowExecution(ctx, *oldExecution) + err = setWorkflowExecution(ctx, *oldExecution, true) if err != nil { log.Printf("Error saving workflow execution actionresult setting: %s", err) return WorkflowExecution{}, fmt.Sprintf("Failed setting workflowexecution actionresult in execution: %s", err), err @@ -2921,7 +2988,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf } for _, trigger := range workflowExecution.Workflow.Triggers { - log.Printf("ID: %s vs %s", trigger.ID, workflowExecution.Start) + log.Printf("[INFO] ID: %s vs %s", trigger.ID, workflowExecution.Start) if trigger.ID == workflowExecution.Start { if trigger.AppName == "User Input" { startFound = true @@ -3024,7 +3091,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf return WorkflowExecution{}, "Failed building missing Docker images", err } - err = setWorkflowExecution(ctx, workflowExecution) + err = setWorkflowExecution(ctx, workflowExecution, true) if err != nil { log.Printf("Error saving workflow execution for updates %s: %s", topic, err) return WorkflowExecution{}, "Failed getting workflowexecution", err @@ -3034,6 +3101,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf // FIXME - add specifics to executionRequest, e.g. specific environment (can run multi onprem) if onpremExecution { // FIXME - tmp name based on future companyname-companyId + // This leads to issues with overlaps. Should set limits and such instead for _, environment := range environments { log.Printf("[INFO] Execution: %s should execute onprem with execution environment \"%s\"", workflowExecution.ExecutionId, environment) @@ -3044,19 +3112,19 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf Environments: environments, } - executionRequestWrapper, err := getWorkflowQueue(ctx, environment) - if err != nil { - executionRequestWrapper = ExecutionRequestWrapper{ - Data: []ExecutionRequest{executionRequest}, - } - } else { - executionRequestWrapper.Data = append(executionRequestWrapper.Data, executionRequest) - } + //executionRequestWrapper, err := getWorkflowQueue(ctx, environment) + //if err != nil { + // executionRequestWrapper = ExecutionRequestWrapper{ + // Data: []ExecutionRequest{executionRequest}, + // } + //} else { + // executionRequestWrapper.Data = append(executionRequestWrapper.Data, executionRequest) + //} //log.Printf("Execution request: %#v", executionRequest) - err = setWorkflowQueue(ctx, executionRequestWrapper, environment) + err = setWorkflowQueue(ctx, executionRequest, environment) if err != nil { - log.Printf("Failed adding to db: %s", err) + log.Printf("[ERROR] Failed adding execution to db: %s", err) } } } @@ -3847,15 +3915,22 @@ func getSpecificWorkflow(resp http.ResponseWriter, request *http.Request) { resp.Write(body) } -func setWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecution) error { +func setWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecution, dbSave bool) error { if len(workflowExecution.ExecutionId) == 0 { log.Printf("Workflowexeciton executionId can't be empty.") return errors.New("ExecutionId can't be empty.") } - key := datastore.NameKey("workflowexecution", workflowExecution.ExecutionId, nil) + cacheKey := fmt.Sprintf("workflowexecution-%s", workflowExecution.ExecutionId) + //requestCache.Delete(cacheKey) + requestCache.Set(cacheKey, &workflowExecution, cache.DefaultExpiration) + if !dbSave { + //log.Printf("[WARNING] SHOULD skip DB saving for execution") + return nil + } // New struct, to not add body, author etc + key := datastore.NameKey("workflowexecution", workflowExecution.ExecutionId, nil) if _, err := dbclient.Put(ctx, key, &workflowExecution); err != nil { log.Printf("Error adding workflow_execution: %s", err) return err @@ -3865,8 +3940,25 @@ func setWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecuti } func getWorkflowExecution(ctx context.Context, id string) (*WorkflowExecution, error) { - key := datastore.NameKey("workflowexecution", strings.ToLower(id), nil) workflowExecution := &WorkflowExecution{} + cacheKey := fmt.Sprintf("workflowexecution-%s", id) + if value, found := requestCache.Get(cacheKey); found { + parsedValue := value.(*WorkflowExecution) + //log.Printf("Found execution for id %s with %d results", parsedValue.ExecutionId, len(parsedValue.Results)) + return parsedValue, nil + + //log.Printf("[INFO] FOUND key %s with value length %d", cacheKey, len(parsedValue)) + //err := json.Unmarshal([]byte(parsedValue), &workflowExecution) + //if err == nil { + // log.Printf("SHOULD RETURN CACHED EXECUTION of length %d", len(parsedValue)) + //} else { + // log.Printf("Failed unmarshalling cached value: %s", err) + //} + } else { + log.Printf("[ERROR] Couldn't find key %s", cacheKey) + } + + key := datastore.NameKey("workflowexecution", strings.ToLower(id), nil) if err := dbclient.Get(ctx, key, workflowExecution); err != nil { return &WorkflowExecution{}, err } diff --git a/backend/tests/execute.sh b/backend/tests/execute.sh index dd8d7a16..5d614f14 100644 --- a/backend/tests/execute.sh +++ b/backend/tests/execute.sh @@ -1,7 +1,26 @@ #!/bin/sh -#curl -XPOST http://localhost:5001/api/v1/workflows/1d9d8ce2-566e-4c3f-8a37-5d6c7d2000b5/execute -d '{"execution_argument":""}' -H "Authorization: Bearer 144308d0-6aab-4d4f-8bb2-75189281ee26" - - +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" +curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" curl -XPOST http://localhost:5001/api/v1/workflows/425efd39-08e7-4390-9387-170c172775f7/execute -d '{"execution_argument":""}' -H "Authorization: Bearer db0373c6-1083-4dec-a05d-3ba73f02ccd4" diff --git a/docker-compose.yml b/docker-compose.yml index 0f7a0666..0ce198d8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,8 +1,8 @@ version: '3' services: frontend: - build: ./frontend - image: ghcr.io/frikky/shuffle-frontend:0.8.45 + #build: ./frontend + image: ghcr.io/frikky/shuffle-frontend:0.8.46 container_name: shuffle-frontend hostname: shuffle-frontend ports: @@ -16,8 +16,8 @@ services: depends_on: - backend backend: - #build: ./backend - image: ghcr.io/frikky/shuffle-backend:0.8.45 + build: ./backend + image: ghcr.io/frikky/shuffle-backend:0.8.46 container_name: shuffle-backend hostname: ${BACKEND_HOSTNAME} # Here for debugging: diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 4a3a2439..fc9bf36b 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -6034,8 +6034,9 @@ const AngularWorkflow = (props) => { //} to_be_copied.replace(" ", "_") - var copyText = document.getElementById("copy_element_shuffle"); - if (copyText !== null) { + const elementName = "copy_element_shuffle" + var copyText = document.getElementById(elementName); + if (copyText !== null && copyText !== null) { navigator.clipboard.writeText(to_be_copied) copyText.select(); copyText.setSelectionRange(0, 99999); /* For mobile devices */ @@ -6044,6 +6045,8 @@ const AngularWorkflow = (props) => { document.execCommand("copy"); alert.success("Copied "+to_be_copied) console.log("COPYING!") + } else { + console.log("Couldn't find element ", elementName) } } diff --git a/functions/onprem/orborus/build.sh b/functions/onprem/orborus/build.sh index 2403b361..4b5bfab2 100644 --- a/functions/onprem/orborus/build.sh +++ b/functions/onprem/orborus/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-orborus -VERSION=0.8.40 +VERSION=0.8.41 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 ea5ac911..4df73a6a 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -115,7 +115,7 @@ func getThisContainerId() { } } else { containerId = "shuffle-orborus" - log.Printf("Failed getting container ID: %s", err) + log.Printf("[WARNING] Failed getting container ID: %s", err) } } @@ -470,7 +470,7 @@ func main() { continue } - log.Printf("Got %d new requests. Executing: %d. Max: %d", len(executionRequests.Data), executionCount, maxConcurrency) + //log.Printf("[INFO] Got %d new requests. Executing: %d. Max: %d", len(executionRequests.Data), executionCount, maxConcurrency) allowed := maxConcurrency - executionCount if len(executionRequests.Data) > allowed {