From ae39da8ffdb55142c822a415f5fd7148545655e1 Mon Sep 17 00:00:00 2001 From: frikky Date: Mon, 16 Nov 2020 18:37:04 +0100 Subject: [PATCH] Fixed most of user input continuations --- backend/go-app/main.go | 131 +++++++++++--- backend/go-app/walkoff.go | 238 +++++++++++++++++-------- frontend/src/views/AngularWorkflow.jsx | 10 +- functions/onprem/orborus/orborus.go | 16 +- functions/onprem/worker/build.sh | 2 +- functions/onprem/worker/worker.go | 122 ++++++++++++- 6 files changed, 400 insertions(+), 119 deletions(-) diff --git a/backend/go-app/main.go b/backend/go-app/main.go index f3b3f789..687016e6 100644 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -74,9 +74,10 @@ var baseAppPath = "/home/frikky/git/shaffuru/tmp/apps" var baseDockerName = "frikky/shuffle" //var syncUrl = "http://192.168.102.54:5002" -//var syncUrl = "http://localhost:5002" var syncUrl = "https://shuffler.io" +//var syncUrl = "http://localhost:5002" + var dbclient *datastore.Client type Userapi struct { @@ -3682,9 +3683,10 @@ func handleNewHook(resp http.ResponseWriter, request *http.Request) { // Let remote endpoint handle access checks (shuffler.io) currentUrl := fmt.Sprintf("https://shuffler.io/api/v1/hooks/webhook_%s", newId) + startNode := requestdata.Start if requestdata.Environment == "cloud" { - log.Printf("[INFO] Should START a cloud webhook for url %s", currentUrl) // https://shuffler.io/v1/hooks/webhook_80184973-3e82-4852-842e-0290f7f34d7c + log.Printf("[INFO] Should START a cloud webhook for url %s for startnode %s", currentUrl, startNode) org, err := getOrg(ctx, user.ActiveOrg.Id) if err != nil { log.Printf("Failed finding org %s: %s", org.Id, err) @@ -3696,7 +3698,7 @@ func handleNewHook(resp http.ResponseWriter, request *http.Request) { Action: "start", OrgId: org.Id, PrimaryItemId: newId, - SecondaryItem: requestdata.Start, + SecondaryItem: startNode, ThirdItem: requestdata.Workflow, } @@ -3713,7 +3715,7 @@ func handleNewHook(resp http.ResponseWriter, request *http.Request) { hook := Hook{ Id: newId, - Start: requestdata.Start, + Start: startNode, Workflows: []string{requestdata.Workflow}, Info: Info{ Name: requestdata.Name, @@ -4709,7 +4711,7 @@ Please contact us at shuffler.io or frikky@shuffler.io if there is an issue with } resp.WriteHeader(200) - resp.Write([]byte("OK")) + resp.Write([]byte(`{"success": true}`)) } func setBadMemcache(ctx context.Context, path string) { @@ -5186,7 +5188,7 @@ func handleNewOutlookRegister(resp http.ResponseWriter, request *http.Request) { } resp.WriteHeader(200) - resp.Write([]byte("OK")) + resp.Write([]byte(`{"success": true}`)) } type OauthToken struct { @@ -6657,22 +6659,34 @@ func handleCloudExecutionOnprem(workflowId, startNode, executionSource, executio if err != nil { return err } + // FIXME: Handle auth _ = workflow - type execStruct struct { - ExecutionSource string `json:"execution_source"` - ExecutionArgument string `json:"execution_argument"` - Start string `json:"start,omitempty"` - } - - parsedArgument := strings.Replace(string(executionArgument), "\"", "\\\"", -1) - newExec := execStruct{ + parsedArgument := executionArgument + newExec := ExecutionRequest{ ExecutionSource: executionSource, ExecutionArgument: parsedArgument, } - //bodyWrapper := fmt.Sprintf(`{"execution_source": "%s", "execution_argument": "%s"}`, executionSource, parsedArgument) + var execution ExecutionRequest + err = json.Unmarshal([]byte(parsedArgument), &execution) + if err == nil { + log.Printf("FOUND EXEC %#v", execution) + if len(execution.ExecutionArgument) > 0 { + parsedArgument := strings.Replace(string(execution.ExecutionArgument), "\\\"", "\"", -1) + log.Printf("New exec argument: %s", execution.ExecutionArgument) + + if strings.HasPrefix(parsedArgument, "{") && strings.HasSuffix(parsedArgument, "}") { + log.Printf("\nData is most likely JSON from %s\n", newExec.ExecutionSource) + } + + newExec.ExecutionArgument = parsedArgument + } + } else { + log.Printf("Unmarshal issue: %s", err) + } + if len(startNode) > 0 { newExec.Start = startNode } @@ -6694,10 +6708,12 @@ func handleCloudExecutionOnprem(workflowId, startNode, executionSource, executio } func handleCloudJob(job CloudSyncJob) error { + // May need authentication in all of these..? + log.Printf("Handle job with type %s and action %s", job.Type, job.Action) if job.Type == "webhook" { if job.Action == "execute" { - log.Printf("Should handle webhook for workflow %s with start node %s and data %s", job.PrimaryItemId, job.SecondaryItem) + log.Printf("Should handle webhook for workflow %s with start node %s and data %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem) err := handleCloudExecutionOnprem(job.PrimaryItemId, job.SecondaryItem, "webhook", job.ThirdItem) if err != nil { log.Printf("Failed executing workflow from cloud hook: %s", err) @@ -6728,14 +6744,79 @@ func handleCloudJob(job CloudSyncJob) error { } } else if job.Type == "user_input" { - if job.Action == "execute" { - log.Printf("Should handle user_input for workflow %s with start node %s and data %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem) - err := handleCloudExecutionOnprem(job.PrimaryItemId, job.SecondaryItem, "user_input", job.ThirdItem) + if job.Action == "continue" { + log.Printf("Should handle user_input CONTINUE for workflow %s with start node %s and execution ID %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem) + // FIXME: Handle authorization + ctx := context.Background() + workflowExecution, err := getWorkflowExecution(ctx, job.ThirdItem) if err != nil { - log.Printf("Failed executing workflow from cloud user_input: %s", err) + return err + } + + if job.PrimaryItemId != workflowExecution.Workflow.ID { + return errors.New("Bad workflow ID when stopping execution.") + } + + workflowExecution.Status = "EXECUTING" + err = setWorkflowExecution(ctx, *workflowExecution) + if err != nil { + return err + } + + fullUrl := fmt.Sprintf("https://shuffler.io/api/v1/workflows/%s/execute?authorization=%s&start=%s&reference_execution=%s&answer=true", job.PrimaryItemId, job.FourthItem, job.SecondaryItem, job.ThirdItem) + newRequest, err := http.NewRequest( + "GET", + fullUrl, + nil, + ) + if err != nil { + log.Printf("Failed continuing workflow in request builder: %s", err) + return err + } + + _, _, err = handleExecution(job.PrimaryItemId, Workflow{}, newRequest) + if err != nil { + log.Printf("Failed continuing workflow from cloud user_input: %s", err) + return err } else { log.Printf("Successfully executed workflow from cloud user_input") } + } else if job.Action == "stop" { + log.Printf("Should handle user_input STOP for workflow %s with start node %s and execution ID %s", job.PrimaryItemId, job.SecondaryItem, job.ThirdItem) + ctx := context.Background() + workflowExecution, err := getWorkflowExecution(ctx, job.ThirdItem) + if err != nil { + return err + } + + if job.PrimaryItemId != workflowExecution.Workflow.ID { + return errors.New("Bad workflow ID when stopping execution.") + } + + /* + if job.FourthItem != workflowExecution.Authorization { + return errors.New("Bad authorization when stopping execution.") + } + */ + + newResults := []ActionResult{} + for _, result := range workflowExecution.Results { + if result.Action.AppName == "User Input" && result.Result == "Waiting for user feedback based on configuration" { + result.Status = "ABORTED" + result.Result = "Aborted manually by user." + } + + newResults = append(newResults, result) + } + + workflowExecution.Results = newResults + workflowExecution.Status = "ABORTED" + err = setWorkflowExecution(ctx, *workflowExecution) + if err != nil { + return err + } + + log.Printf("Successfully updated user input to aborted.") } } else { log.Printf("No handler for type %s and action %s", job.Type, job.Action) @@ -6763,7 +6844,7 @@ func remoteOrgJobController(org Org, body []byte) error { log.Printf("Should stop org job controller") if strings.Contains(responseData.Reason, "Bad apikey") { - log.Printf("Bad apikey. Stopping sync for org!") + log.Printf("Bad apikey. Stopping sync for org?: %s", responseData.Reason) if value, exists := scheduledOrgs[org.Id]; exists { // Looks like this does the trick? Hurr @@ -6786,7 +6867,7 @@ func remoteOrgJobController(org Org, body []byte) error { log.Printf("Successfully updated the org to not sync") } - return errors.New("Stopped schedule for org because of bad apikey.") + return errors.New("Stopped schedule for org locally because of bad apikey.") } else { return errors.New(fmt.Sprintf("Failed finding the schedule for org %s", org.Id)) } @@ -7223,7 +7304,7 @@ func runInit(ctx context.Context) { } //interval := int(org.SyncConfig.Interval) - interval := 5 + interval := 15 if interval == 0 { log.Printf("Skipping org %s because sync isn't set (0).", org.Id) continue @@ -7731,7 +7812,7 @@ func handleCloudSetup(resp http.ResponseWriter, request *http.Request) { // return //} - log.Printf("Apidata: %s", tmpData.Apikey) + //log.Printf("Apidata: %s", tmpData.Apikey) // FIXME: Path client := &http.Client{} @@ -7808,7 +7889,7 @@ func handleCloudSetup(resp http.ResponseWriter, request *http.Request) { return } - log.Printf("Respbody: %s", string(respBody)) + //log.Printf("Respbody: %s", string(respBody)) responseData := retStruct{} err = json.Unmarshal(respBody, &responseData) if err != nil { diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 0b95ed20..729fbdd8 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -56,15 +56,15 @@ var scheduledOrgs = map[string]*newscheduler.Job{} //} type ExecutionRequest struct { - ExecutionId string `json:"execution_id"` - ExecutionArgument string `json:"execution_argument"` - ExecutionSource string `json:"execution_source"` - WorkflowId string `json:"workflow_id"` - Environments []string `json:"environments"` - Authorization string `json:"authorization"` - Status string `json:"status"` - Start string `json:"start"` - Type string `json:"type"` + ExecutionId string `json:"execution_id,omitempty"` + ExecutionArgument string `json:"execution_argument,omitempty"` + ExecutionSource string `json:"execution_source,omitempty"` + WorkflowId string `json:"workflow_id,omitempty"` + Environments []string `json:"environments,omitempty"` + Authorization string `json:"authorization,omitempty"` + Status string `json:"status,omitempty"` + Start string `json:"start,omitempty"` + Type string `json:"type,omitempty"` } type SyncFeatures struct { @@ -685,7 +685,7 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque //} resp.WriteHeader(200) - resp.Write([]byte("OK")) + resp.Write([]byte(`{"success": true}`)) } // FIXME: Authenticate this one? Can org ID be auth enough? @@ -899,6 +899,48 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { } } + if actionResult.Status == "WAITING" && actionResult.Action.AppName == "User Input" { + log.Printf("SHOULD WAIT A BIT AND RUN CLOUD STUFF WITH USER INPUT! WAITING!") + + var trigger Trigger + err = json.Unmarshal([]byte(actionResult.Result), &trigger) + if err != nil { + log.Printf("Failed unmarshaling actionresult for user input: %s", err) + resp.WriteHeader(401) + resp.Write([]byte(`{"success": false}`)) + return + } + + orgId := workflowExecution.ExecutionOrg + if len(workflowExecution.OrgId) == 0 && len(workflowExecution.Workflow.OrgId) > 0 { + orgId = workflowExecution.Workflow.OrgId + } + + err := handleUserInput(trigger, orgId, workflowExecution.Workflow.ID, workflowExecution.ExecutionId) + if err != nil { + log.Printf("Failed userinput handler: %s", err) + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Error: %s"}`, err))) + } else { + log.Printf("Successful userinput handler") + resp.WriteHeader(200) + resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "CLOUD IS DONE"}`))) + + actionResult.Result = "Waiting for user feedback based on configuration" + + workflowExecution.Results = append(workflowExecution.Results, actionResult) + workflowExecution.Status = actionResult.Status + err = setWorkflowExecution(ctx, *workflowExecution) + if err != nil { + log.Printf("Failed ") + } else { + log.Printf("Successfully set the execution to waiting.") + } + } + + return + } + if actionResult.Status == "ABORTED" || actionResult.Status == "FAILURE" { log.Printf("Actionresult is %s. Should set workflowExecution and exit all running functions", actionResult.Status) @@ -977,7 +1019,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { for _, result := range workflowExecution.Results { if result.Status == "EXECUTING" { result.Status = actionResult.Status - result.Result = "Aborted because of an unknown error" + result.Result = "Aborted because of error in another node" } if len(result.Result) > 0 { @@ -1201,10 +1243,10 @@ func handleExecutionStatistics(execution WorkflowExecution) { for _, result := range execution.Results { resultCheck := JSONCheck(result.Result) if !resultCheck { - log.Printf("Result is NOT JSON!") + //log.Printf("Result is NOT JSON!") continue } else { - log.Printf("Result IS JSON!") + //log.Printf("Result IS JSON!") } @@ -1839,6 +1881,13 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) { } } + if len(triggerType) == 0 { + log.Printf("No type specified for user input node") + resp.WriteHeader(401) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No contact option specified in user input"}`))) + return + } + // FIXME: This is not the right time to send them, BUT it's well served for testing. Save -> send email / sms _ = triggerInformation if strings.Contains(triggerType, "email") { @@ -1849,38 +1898,6 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) { return } - // triggerid, start node, workflowid, argument - /* - startNode := "" - referenceExecutionId := "" - action := CloudSyncJob{ - Type: "user_input", - Action: "send_email", - OrgId: user.ActiveOrg.Id, - PrimaryItemId: workflow.ID, - SecondaryItem: startNode, - ThirdItem: triggerInformation, - FourthItem: email, - FifthItem: referenceExecutionId, - } - - org, err := getOrg(ctx, user.ActiveOrg.Id) - if err != nil { - log.Printf("Failed email send to cloud", err) - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) - return - } - - err = executeCloudAction(action, org.SyncConfig.Apikey) - if err != nil { - log.Printf("Failed email send to cloud", err) - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) - return - } - */ - log.Printf("Should send email to %s during execution.", email) } if strings.Contains(triggerType, "sms") { @@ -1891,37 +1908,6 @@ func saveWorkflow(resp http.ResponseWriter, request *http.Request) { return } - /* - startNode := "" - referenceExecutionId := "" - action := CloudSyncJob{ - Type: "user_input", - Action: "send_sms", - OrgId: user.ActiveOrg.Id, - PrimaryItemId: workflow.ID, - SecondaryItem: startNode, - ThirdItem: triggerInformation, - FourthItem: sms, - FifthItem: referenceExecutionId, - } - - org, err := getOrg(ctx, user.ActiveOrg.Id) - if err != nil { - log.Printf("Failed email send to cloud", err) - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) - return - } - - err = executeCloudAction(action, org.SyncConfig.Apikey) - if err != nil { - log.Printf("Failed email send to cloud", err) - resp.WriteHeader(401) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err))) - return - } - */ - log.Printf("Should send SMS to %s during execution.", sms) } } @@ -2346,7 +2332,7 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) { return } } else { - log.Printf("API key %s is correct to abort %s", parsedKey, executionId) + log.Printf("API key to abort/finish execution %s is correct.", executionId) } if workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" || workflowExecution.Status == "FINISHED" { @@ -2367,7 +2353,7 @@ func abortExecution(resp http.ResponseWriter, request *http.Request) { for _, result := range workflowExecution.Results { if result.Status == "EXECUTING" { result.Status = "ABORTED" - result.Result = "Aborted because of an unknown error" + result.Result = "Aborted because of error in another node" } if len(result.Result) > 0 { @@ -2446,7 +2432,7 @@ func cleanupExecutions(resp http.ResponseWriter, request *http.Request) { } resp.WriteHeader(200) - resp.Write([]byte("OK")) + resp.Write([]byte(`{"success": true}`)) } func handleExecution(id string, workflow Workflow, request *http.Request) (WorkflowExecution, string, error) { @@ -2507,6 +2493,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf } // This one doesn't really matter. + log.Printf("Running POST execution with data %s", body) var execution ExecutionRequest err = json.Unmarshal(body, &execution) if err != nil { @@ -2797,6 +2784,16 @@ 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) + if trigger.ID == workflowExecution.Start { + if trigger.AppName == "User Input" { + startFound = true + break + } + } + } + if !startFound { log.Printf("Startnode %s doesn't exist!", workflowExecution.Start) return WorkflowExecution{}, fmt.Sprintf("Workflow action %s doesn't exist in workflow", workflowExecution.Start), errors.New(fmt.Sprintf(`Workflow start node "%s" doesn't exist. Exiting!`, workflowExecution.Start)) @@ -3509,6 +3506,7 @@ func scheduleWorkflow(resp http.ResponseWriter, request *http.Request) { SecondaryItem: schedule.Frequency, ThirdItem: workflow.ID, FourthItem: schedule.ExecutionArgument, + FifthItem: startNode, } timeNow := int64(time.Now().Unix()) @@ -6128,3 +6126,85 @@ func removeOutlookTriggerFunction(ctx context.Context, triggerId string) error { _ = resp return nil } + +func handleUserInput(trigger Trigger, organizationId string, workflowId string, referenceExecution string) error { + // E.g. check email + sms := "" + email := "" + triggerType := "" + triggerInformation := "" + for _, item := range trigger.Parameters { + if item.Name == "alertinfo" { + triggerInformation = item.Value + } else if item.Name == "type" { + triggerType = item.Value + } else if item.Name == "email" { + email = item.Value + } else if item.Name == "sms" { + sms = item.Value + } + } + + if len(triggerType) == 0 { + log.Printf("No type specified for user input node") + return errors.New("No type specified for user input node") + } + + // FIXME: This is not the right time to send them, BUT it's well served for testing. Save -> send email / sms + ctx := context.Background() + startNode := trigger.ID + if strings.Contains(triggerType, "email") { + action := CloudSyncJob{ + Type: "user_input", + Action: "send_email", + OrgId: organizationId, + PrimaryItemId: workflowId, + SecondaryItem: startNode, + ThirdItem: triggerInformation, + FourthItem: email, + FifthItem: referenceExecution, + } + + org, err := getOrg(ctx, organizationId) + if err != nil { + log.Printf("Failed email send to cloud: %s", err) + return err + } + + err = executeCloudAction(action, org.SyncConfig.Apikey) + if err != nil { + log.Printf("Failed email send to cloud", err) + return err + } + + log.Printf("Should send email to %s during execution.", email) + } + if strings.Contains(triggerType, "sms") { + action := CloudSyncJob{ + Type: "user_input", + Action: "send_sms", + OrgId: organizationId, + PrimaryItemId: workflowId, + SecondaryItem: startNode, + ThirdItem: triggerInformation, + FourthItem: sms, + FifthItem: referenceExecution, + } + + org, err := getOrg(ctx, organizationId) + if err != nil { + log.Printf("Failed email send to cloud", err) + return err + } + + err = executeCloudAction(action, org.SyncConfig.Apikey) + if err != nil { + log.Printf("Failed email send to cloud", err) + return err + } + + log.Printf("Should send SMS to %s during execution.", sms) + } + + return nil +} diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 2f10d25e..ba4d6199 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -478,7 +478,7 @@ const AngularWorkflow = (props) => { } } - if (responseJson.status === "ABORTED" || responseJson.status === "STOPPED" || responseJson.status === "FAILURE") { + if (responseJson.status === "ABORTED" || responseJson.status === "STOPPED" || responseJson.status === "FAILURE" || responseJson.status == "WAITING") { stop() setExecutionRunning(false) @@ -5195,12 +5195,12 @@ const AngularWorkflow = (props) => { />
- -
{executionData.results === undefined || executionData.results === null || executionData.results.length === 0 && executionData.status === "EXECUTING" ? @@ -5719,7 +5719,7 @@ const AngularWorkflow = (props) => { const statusColor = data.status === "FINISHED" || data.status === "SUCCESS" ? "green" : data.status === "ABORTED" || data.status === "FAILURE" ? "red" : "orange" const actionimg = curapp === null ? null : - {data.action.app_name} + {data.action.app_name} return (
diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 2520377e..412f0263 100644 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -145,7 +145,21 @@ func deployWorker(image string, identifier string, env []string) { if err != nil { log.Printf("[ERROR] Container create error: %s", err) - return + + identifier := fmt.Sprintf("%s-new", identifier) + cont, err = dockercli.ContainerCreate( + context.Background(), + config, + hostConfig, + nil, + nil, + identifier, + ) + + if err != nil { + log.Printf("[ERROR] Container create error(2): %s", err) + return + } } err = dockercli.ContainerStart(context.Background(), cont.ID, types.ContainerStartOptions{}) diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 3d2f9ebe..5872707f 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -2,7 +2,7 @@ NAME=worker VERSION=0.8.0 echo "Running docker build with $NAME:$VERSION" -CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin . +#CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin . docker build . -t frikky/shuffle:$NAME -t docker.pkg.github.com/frikky/shuffle/$NAME:$VERSION -t ghcr.io/frikky/$NAME:$VERSION # Push both for now.. diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index c6b3889a..244ecc4b 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -236,10 +236,9 @@ type Action struct { AuthNotRequired bool `json:"auth_not_required" datastore:"auth_not_required" yaml:"auth_not_required"` } -// Added environment for location to execute type Trigger struct { AppName string `json:"app_name" datastore:"app_name"` - Description string `json:"description" datastore:"description"` + Description string `json:"description" datastore:"description,noindex"` LongDescription string `json:"long_description" datastore:"long_description"` Status string `json:"status" datastore:"status"` AppVersion string `json:"app_version" datastore:"app_version"` @@ -253,6 +252,7 @@ type Trigger struct { Environment string `json:"environment" datastore:"environment"` TriggerType string `json:"trigger_type" datastore:"trigger_type"` Name string `json:"name" datastore:"name"` + Tags []string `json:"tags" datastore:"tags" yaml:"tags"` Parameters []WorkflowAppActionParameter `json:"parameters" datastore: "parameters,noindex"` Position struct { X float64 `json:"x" datastore:"x"` @@ -549,6 +549,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W // source = parent node, dest = child node // parent can have more children, child can have more parents + extra := 0 for _, branch := range workflowExecution.Workflow.Branches { // Check what the parent is first. If it's trigger - skip sourceFound := false @@ -563,6 +564,21 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W } } + for _, trigger := range workflowExecution.Workflow.Triggers { + if trigger.AppName != "User Input" { + continue + } + + if trigger.ID == branch.SourceID { + sourceFound = true + extra += 1 + } + + if trigger.ID == branch.DestinationID { + destinationFound = true + } + } + if sourceFound { parents[branch.DestinationID] = append(parents[branch.DestinationID], branch.SourceID) } else { @@ -576,7 +592,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W } } - log.Printf("Actions: %d", len(workflowExecution.Workflow.Actions)) + log.Printf("Actions: %d + Special Triggers: %d", len(workflowExecution.Workflow.Actions), extra) for _, action := range workflowExecution.Workflow.Actions { if action.Environment != environment { continue @@ -775,7 +791,41 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W // IF NOT VISITED && IN toExecuteOnPrem // SKIP if it's not onprem for _, nextAction := range nextActions { - action := getAction(workflowExecution, nextAction) + action := getAction(workflowExecution, nextAction, environment) + if action.AppName == "User Input" { + log.Printf("USER INPUT!") + + if action.ID == workflowExecution.Start { + log.Printf("Skipping because it's the startnode") + } else { + log.Printf("Should stop after this iteration because it's user-input based. %#v", action) + trigger := Trigger{} + for _, innertrigger := range workflowExecution.Workflow.Triggers { + if innertrigger.ID == action.ID { + trigger = innertrigger + break + } + } + + trigger.LargeImage = "" + triggerData, err := json.Marshal(trigger) + if err != nil { + log.Printf("Failed unmarshalling action: %s", err) + triggerData = []byte("Failed unmarshalling. Cancel execution!") + } + + err = runUserInput(client, action, workflowExecution.Workflow.ID, workflowExecution.ExecutionId, workflowExecution.Authorization, string(triggerData)) + if err != nil { + log.Printf("Failed launching backend magic: %s", err) + os.Exit(3) + } else { + log.Printf("Launched user input node succesfully!") + os.Exit(3) + } + + break + } + } // check visited and onprem if arrayContains(visited, nextAction) { @@ -911,7 +961,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W // https://devblogs.microsoft.com/oldnewthing/20100203-00/?p=15083 maxSize := 32700 - len(string(actionData)) - 2000 if len(executionData) < maxSize { - log.Printf("ADDING FULL_EXECUTION because size is larger than %d", maxSize) + log.Printf("ADDING FULL_EXECUTION because size is smaller than %d", maxSize) env = append(env, fmt.Sprintf("FULL_EXECUTION=%s", string(executionData))) } else { log.Printf("Skipping FULL_EXECUTION because size is larger than %d", maxSize) @@ -973,7 +1023,7 @@ func handleExecution(client *http.Client, req *http.Request, workflowExecution W shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) } - log.Printf("Status: %s, Results: %d, actions: %d", workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) + log.Printf("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) shutdown(workflowExecution.ExecutionId, workflowExecution.Workflow.ID) @@ -1075,16 +1125,73 @@ func getResult(workflowExecution WorkflowExecution, id string) ActionResult { return ActionResult{} } -func getAction(workflowExecution WorkflowExecution, id string) Action { +func getAction(workflowExecution WorkflowExecution, id, environment string) Action { for _, action := range workflowExecution.Workflow.Actions { if action.ID == id { return action } } + for _, trigger := range workflowExecution.Workflow.Triggers { + if trigger.ID == id { + return Action{ + ID: trigger.ID, + AppName: trigger.AppName, + Name: trigger.AppName, + Environment: environment, + } + log.Printf("FOUND TRIGGER: %#v!", trigger) + } + } + return Action{} } +func runUserInput(client *http.Client, action Action, workflowId, workflowExecutionId, authorization string, configuration string) error { + timeNow := time.Now().Unix() + result := ActionResult{ + Action: action, + ExecutionId: workflowExecutionId, + Authorization: authorization, + Result: configuration, + StartedAt: timeNow, + CompletedAt: 0, + Status: "WAITING", + } + + 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] Body: %s", string(body)) + return nil +} + func runTestExecution(client *http.Client, workflowId, apikey string) (string, string) { fullUrl := fmt.Sprintf("%s/api/v1/workflows/%s/execute", baseUrl, workflowId) req, err := http.NewRequest( @@ -1173,7 +1280,6 @@ func main() { shutdown(executionId, "") } - // FIXME - tmp data := fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization) fullUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl) req, err := http.NewRequest(