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) => {
/>
-
+ 
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(