From a7e0b2706b6efac939e5dc4380dc88f27aeee72d Mon Sep 17 00:00:00 2001 From: frikky Date: Sun, 3 Jan 2021 16:45:09 +0100 Subject: [PATCH] FEATURE: Added new subflow trigger --- functions/onprem/worker/worker.go | 161 ++++++++++++++++++++++++++++-- 1 file changed, 152 insertions(+), 9 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index d0267d14..9df5fea0 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -926,6 +926,106 @@ func runFilter(workflowExecution WorkflowExecution, action Action) { } +func handleSubworkflowExecution(client *http.Client, workflowExecution WorkflowExecution, action Trigger, baseAction Action) error { + apikey := "" + workflowId := "" + executionArgument := "" + for _, parameter := range action.Parameters { + log.Printf("Parameter name: %s", parameter.Name) + if parameter.Name == "user_apikey" { + apikey = parameter.Value + } else if parameter.Name == "workflow" { + workflowId = parameter.Value + } else if parameter.Name == "data" { + executionArgument = parameter.Value + } + } + + //handleSubworkflowExecution(workflowExecution, action) + status := "SUCCESS" + baseResult := `{"success": true}` + if len(apikey) == 0 || len(workflowId) == 0 { + status = "FAILURE" + baseResult = `{"success": false}` + } else { + log.Printf("Should execute workflow %s with APIKEY %s and data %s", workflowId, apikey, executionArgument) + fullUrl := fmt.Sprintf("%s/api/workflows/%s/execute", baseUrl, workflowId) + req, err := http.NewRequest( + "POST", + fullUrl, + bytes.NewBuffer([]byte(executionArgument)), + ) + + if err != nil { + log.Printf("Error building test request: %s", err) + return err + } + + req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", apikey)) + newresp, err := client.Do(req) + if err != nil { + log.Printf("Error running test request: %s", err) + return err + } + + body, err := ioutil.ReadAll(newresp.Body) + if err != nil { + log.Printf("Failed reading body when waiting: %s", err) + return err + } + + log.Printf("Execution Result: %s", body) + } + + timeNow := time.Now().Unix() + result := ActionResult{ + Action: baseAction, + ExecutionId: workflowExecution.ExecutionId, + Authorization: workflowExecution.Authorization, + Result: baseResult, + StartedAt: timeNow, + CompletedAt: 0, + Status: status, + } + + resultData, err := json.Marshal(result) + if err != nil { + return err + } + + fullUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl) + req, err := http.NewRequest( + "POST", + fullUrl, + bytes.NewBuffer([]byte(resultData)), + ) + + if err != nil { + log.Printf("Error building test request: %s", err) + return err + } + + newresp, err := client.Do(req) + if err != nil { + log.Printf("Error running test request: %s", err) + return err + } + + body, err := ioutil.ReadAll(newresp.Body) + if err != nil { + log.Printf("Failed reading body when waiting: %s", err) + return err + } + + log.Printf("[INFO] Subworkflow Body: %s", string(body)) + + if status == "FAILURE" { + return errors.New("[ERROR] Failed to execute subworkflow") + } else { + return nil + } +} + func handleExecution(client *http.Client, req *http.Request, workflowExecution WorkflowExecution) error { // if no onprem runs (shouldn't happen, but extra check), exit // if there are some, load the images ASAP for the app @@ -965,10 +1065,10 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W } for _, trigger := range workflowExecution.Workflow.Triggers { - if trigger.AppName != "User Input" { + log.Printf("Appname trigger: %s", trigger.AppName) + if !(trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow") { continue } - if trigger.ID == branch.SourceID { sourceFound = true extra += 1 @@ -976,6 +1076,10 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W if trigger.ID == branch.DestinationID { destinationFound = true + + if !sourceFound { + extra += 1 + } } } @@ -1198,7 +1302,45 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W continue } - if action.AppName == "User Input" { + if action.AppName == "Shuffle Workflow" { + log.Printf("SHUFFLE WORKFLOW: %#v", action) + action.Environment = environment + action.AppName = "shuffle-subflow" + action.Name = "run_subflow" + action.AppVersion = "1.0.0" + + //appname := action.AppName + //appversion := action.AppVersion + //appname = strings.Replace(appname, ".", "-", -1) + //appversion = strings.Replace(appversion, ".", "-", -1) + // shuffle-subflow_1.0.0 + + //visited = append(visited, action.ID) + //executed = append(executed, action.ID) + + trigger := Trigger{} + for _, innertrigger := range workflowExecution.Workflow.Triggers { + if innertrigger.ID == action.ID { + trigger = innertrigger + break + } + } + + action.Parameters = []WorkflowAppActionParameter{} + for _, parameter := range trigger.Parameters { + parameter.Variant = "STATIC_VALUE" + action.Parameters = append(action.Parameters, parameter) + } + + //trigger.LargeImage = "" + //err = handleSubworkflowExecution(client, workflowExecution, trigger, action) + //if err != nil { + // log.Printf("[ERROR] Failed to execute subworkflow: %s", err) + //} else { + // log.Printf("[INFO] Executed subworkflow!") + //} + //continue + } else if action.AppName == "User Input" { log.Printf("USER INPUT!") if action.ID == workflowExecution.Start { @@ -1466,19 +1608,19 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W err = json.Unmarshal(body, &workflowExecution) if err != nil { - log.Printf("Failed workflowExecution unmarshal: %s", err) + log.Printf("[ERROR] Failed workflowExecution unmarshal: %s", err) time.Sleep(time.Duration(sleepTime) * time.Second) continue } if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" { - log.Printf("Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId) + log.Printf("[INFO] Workflow %s is finished. Exiting worker.", workflowExecution.ExecutionId) shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) } - log.Printf("Status: %s, Results: %d, actions: %d", workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra) + log.Printf("[INFO] Status: %s, Results: %d, actions: %d", workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra) if workflowExecution.Status != "EXECUTING" { - log.Printf("Exiting as worker execution has status %s!", workflowExecution.Status) + log.Printf("[WARNING] Exiting as worker execution has status %s!", workflowExecution.Status) shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) } @@ -1592,6 +1734,7 @@ func getAction(workflowExecution WorkflowExecution, id, environment string) Acti AppName: trigger.AppName, Name: trigger.AppName, Environment: environment, + Label: trigger.Label, } log.Printf("FOUND TRIGGER: %#v!", trigger) } @@ -1641,7 +1784,7 @@ func runUserInput(client *http.Client, action Action, workflowId, workflowExecut return err } - log.Printf("[INFO] Body: %s", string(body)) + log.Printf("[INFO] User Input Body: %s", string(body)) return nil } @@ -1671,7 +1814,7 @@ func runTestExecution(client *http.Client, workflowId, apikey string) (string, s return "", "" } - log.Printf("[INFO] Body: %s", string(body)) + log.Printf("[INFO] Test Body: %s", string(body)) var workflowExecution WorkflowExecution err = json.Unmarshal(body, &workflowExecution) if err != nil {