Added distributed possibility for memcached. Not optimal, but required for HA.

Author: frikky

db-connector.go

@@ -8330,7 +8330,6 @@ func SetWorkflow(ctx context.Context, workflow Workflow, id string, optionalEdit
 		}
 
 		// Find the key for "workflows_<workflow.org_id>" and update the cache for this one. If it doesn't exist, add it
-
 		// Get the cache for the workflows
 		cacheKey = fmt.Sprintf("%s_workflows", workflow.OrgId)
 		cache, err := GetCache(ctx, cacheKey)
@@ -11999,6 +11998,21 @@ func RunInit(dbclient datastore.Client, storageClient storage.Client, gceProject
 	// In case of downtime / large requests
 	if len(memcached) > 0 {
 		mc.Timeout = 10 * time.Second
+
+
+		if strings.Contains(memcached, ",") {
+
+			newMemcached := []string{}
+			for _, memcached := range strings.Split(memcached, ",") {
+				memcached = strings.TrimSpace(memcached)
+				if len(memcached) > 0 {
+					newMemcached = append(newMemcached, memcached)
+				}
+			}
+
+			log.Printf("[DEBUG] Multiple memcached servers detected. Split into %#v", newMemcached)
+			mc = gomemcache.New(newMemcached...)
+		}
 	}
 
 	requestCache = cache.New(35*time.Minute, 35*time.Minute)
@@ -13008,7 +13022,7 @@ func ValidateFinished(ctx context.Context, extra int, workflowExecution Workflow
 
 	workflowExecution, _ = Fixexecution(ctx, workflowExecution)
 	//if rand.Intn(5) == 1 || len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions) {
-	log.Printf("[INFO][%s] Workflow Validation. Status: %s, Actions: %d, Extra: %d, Results: %d\n", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results))
+	log.Printf("[INFO][%s] Workflow Finished Check. Status: %s, Actions: %d, Extra: %d, Results: %d\n", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Workflow.Actions), extra, len(workflowExecution.Results))
 	if len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions)+extra && len(workflowExecution.Workflow.Actions) > 0 {
 		validResults := 0
 		invalidResults := 0

shared.go

@@ -22474,35 +22474,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h
 								})
 							}
 
-							// Commented out as we don't want to stop the app, but just continue with the old tokens
-							/*
-								actionRes := ActionResult{
-									Action:        action,
-									ExecutionId:   workflowExecution.ExecutionId,
-									Authorization: workflowExecution.Authorization,
-									Result:        fmt.Sprintf(`{"success": false, "reason": "Failed running oauth2 request to refresh oauth2 tokens for this app."}`),
-									StartedAt:     workflowExecution.StartedAt,
-									CompletedAt:   workflowExecution.StartedAt,
-									Status:        "FAILURE",
-								}
-
-								workflowExecution.Results = append(workflowExecution.Results, actionRes)
-								cacheData := []byte("1")
-
-								newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
-								err = SetCache(ctx, newExecId, cacheData, 2)
-								if err != nil {
-									log.Printf("[WARNING] Failed setting base cache for failed Oauth2 action %s: %s", newExecId, err)
-								}
-
-								b, err := json.Marshal(actionRes)
-								if err == nil {
-									err = SetCache(ctx, fmt.Sprintf("%s_result", newExecId), b, 2)
-									if err != nil {
-										log.Printf("[WARNING] Failed setting result cache for failed Oauth2 action %s: %s", newExecId, err)
-									}
-								}
-							*/
+							// FIXME: There used to be code here to stop the app, but for now we just continue with the old tokens
 						}
 
 						allAuths[authIndex] = newAuth
@@ -30437,6 +30409,8 @@ func checkExecutionStatus(ctx context.Context, exec *WorkflowExecution) *Workflo
 
 	workflow.Validation.TotalProblems = len(workflow.Validation.Errors) + len(workflow.Validation.SubflowApps)
 
+	log.Printf("\n\n\nVALIDATION RUNNING\n\n\n")
+
 	// Updating the workflow to show the right status every time for now
 	workflowChanged = true
 	workflow.Validation.ValidationRan = true