From faa555e26f3e622531dcf10892cabdddbc4915de Mon Sep 17 00:00:00 2001 From: frikky Date: Fri, 27 Aug 2021 03:17:15 +0200 Subject: [PATCH] #223: Fixed faster default responses from subflows --- backend/app_sdk/app_base.py | 23 +++++ backend/go-app/go.mod | 2 +- backend/go-app/main.go | 2 +- backend/go-app/walkoff.go | 123 +++++-------------------- frontend/src/views/AngularWorkflow.jsx | 1 + frontend/src/views/Workflows.jsx | 73 ++++++++++++--- functions/onprem/worker/build.sh | 2 +- functions/onprem/worker/go.mod | 4 +- functions/onprem/worker/worker.go | 119 +++++++++++++----------- 9 files changed, 178 insertions(+), 171 deletions(-) diff --git a/backend/app_sdk/app_base.py b/backend/app_sdk/app_base.py index a6871a4f..101569fa 100644 --- a/backend/app_sdk/app_base.py +++ b/backend/app_sdk/app_base.py @@ -1714,6 +1714,29 @@ class AppBase: elif check.lower() == "contains": if destinationvalue.lower() in sourcevalue.lower(): return True + + elif check.lower() == "is empty": + if len(sourcevalue) == 0: + return True + + if str(sourcevalue) == 0: + return True + + return False + #if tmp == "[]": + # tmp = [] + + #if type(tmp) == list and len(tmp) == 0 and not flip: + # new_list.append(item) + #elif type(tmp) == list and len(tmp) > 0 and flip: + # new_list.append(item) + #elif type(tmp) == str and not tmp and not flip: + # new_list.append(item) + #elif type(tmp) == str and tmp and flip: + # new_list.append(item) + #else: + # failed_list.append(item) + elif check.lower() == "contains_any_of": newvalue = [destinationvalue.lower()] if "," in destinationvalue: diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index fe726c85..f7fe10af 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -2,7 +2,7 @@ module shuffle go 1.13 -//replace github.com/frikky/shuffle-shared => ../../../../git/shuffle-shared +replace github.com/frikky/shuffle-shared => ../../../../git/shuffle-shared //replace github.com/frikky/kin-openapi => ../../../../git/kin-openapi diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 24404c6c..43852d29 100644 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -5614,7 +5614,7 @@ func makeWorkflowPublic(resp http.ResponseWriter, request *http.Request) { if workflow.OrgId == user.ActiveOrg.Id && user.Role == "admin" { log.Printf("[AUDIT] User %s is accessing workflow %s as admin (public)", user.Username, workflow.ID) } else { - log.Printf("[WARNING] Wrong user (%s) for workflow %s (public)", user.Username, workflow.ID) + log.Printf("[AUDIT] Wrong user (%s) for workflow %s (public)", user.Username, workflow.ID) resp.WriteHeader(401) resp.Write([]byte(`{"success": false}`)) return diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 205dea93..84562ff4 100644 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -628,7 +628,7 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque 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) + //log.Printf("[INFO] Deleted %d keys from org %s", len(ids), parsedId) } //var newExecutionRequests ExecutionRequestWrapper @@ -687,7 +687,7 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { if len(executionRequests.Data) == 0 { executionRequests.Data = []shuffle.ExecutionRequest{} } else { - log.Printf("[INFO] Executionrequests (%s): %d", id, len(executionRequests.Data)) + //log.Printf("[INFO] Executionrequests (%s): %d", id, len(executionRequests.Data)) //log.Printf("IDS: %#v", executionRequests.Data[0].ExecutionId) } @@ -755,80 +755,6 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) { } -// Checks if data is sent from Worker >0.8.51, which sends a full execution -// instead of individial results -func validateNewWorkerExecution(body []byte) error { - ctx := context.Background() - var execution shuffle.WorkflowExecution - err := json.Unmarshal(body, &execution) - if err != nil { - log.Printf("[WARNING] Failed execution unmarshaling: %s", err) - return err - } - //log.Printf("\n\nGOT EXEC WITH RESULT %#v (%d)\n\n", execution.Status, len(execution.Results)) - - baseExecution, err := shuffle.GetWorkflowExecution(ctx, execution.ExecutionId) - if err != nil { - log.Printf("[ERROR] Failed getting execution (workflowqueue) %s: %s", execution.ExecutionId, err) - return err - } - - if baseExecution.Authorization != execution.Authorization { - return errors.New("Bad authorization when validating execution") - } - - // used to validate if it's actually the right marshal - if len(baseExecution.Workflow.Actions) != len(execution.Workflow.Actions) { - return errors.New(fmt.Sprintf("Bad length of actions (probably normal app): %d", len(execution.Workflow.Actions))) - } - - if len(baseExecution.Workflow.Triggers) != len(execution.Workflow.Triggers) { - return errors.New(fmt.Sprintf("Bad length of trigger: %d (probably normal app)", len(execution.Workflow.Triggers))) - } - - if len(baseExecution.Results) >= len(execution.Results) { - return errors.New(fmt.Sprintf("Can't have less actions in a full execution than what exists: %d (old) vs %d (new)", len(baseExecution.Results), len(execution.Results))) - } - - //if baseExecution.Status != "WAITING" && baseExecution.Status != "EXECUTING" { - // return errors.New(fmt.Sprintf("Workflow is already finished or failed. Can't update")) - //} - - if execution.Status == "EXECUTING" { - //log.Printf("[INFO] Inside executing.") - extra := 0 - for _, trigger := range execution.Workflow.Triggers { - //log.Printf("Appname trigger (0): %s", trigger.AppName) - if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" { - extra += 1 - } - } - - if len(execution.Workflow.Actions)+extra == len(execution.Results) { - execution.Status = "FINISHED" - } - - //log.Printf("[INFO] BASEEXECUTION LENGTH: %d", len(baseExecution.Workflow.Actions)+extra) - } - - // FIXME: Add extra here - //executionLength := len(baseExecution.Workflow.Actions) - //if executionLength != len(execution.Results) { - // return errors.New(fmt.Sprintf("Bad length of actions vs results: want: %d have: %d", executionLength, len(execution.Results))) - //} - - //log.Printf("\n\nSHOULD SET BACKEND DATA FOR EXEC \n\n") - err = shuffle.SetWorkflowExecution(ctx, execution, true) - if err == nil { - log.Printf("[INFO] Set workflowexecution based on new worker (>0.8.53) for execution %s. Actions: %d, Triggers: %d, Results: %d, Status: %s", execution.ExecutionId, len(execution.Workflow.Actions), len(execution.Workflow.Triggers), len(execution.Results), execution.Status) //, execution.Result) - //log.Printf("[INFO] Successfully set the execution to wait.") - } else { - log.Printf("[WARNING] Failed to set the execution to wait.") - } - - return nil -} - func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { cors := handleCors(resp, request) if cors { @@ -844,7 +770,7 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { } //log.Printf("Actionresult unmarshal: %s", string(body)) - err = validateNewWorkerExecution(body) + err = shuffle.ValidateNewWorkerExecution(body) if err == nil { resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Success"}`))) @@ -980,31 +906,9 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl return } - //log.Printf("NEW LENGTH: %d", len(workflowExecution.Results)) - _ = dbSave //resultLength := len(workflowExecution.Results) setExecution := true - - //newExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId) - //if err == nil { - // //log.Printf("GOT GOOD EXECUTION CACHE FOR %s!", workflowExecution.ExecutionId) - // if len(newExecution.Results) > 0 && len(newExecution.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) - // } - - // attempts += 1 - // //if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) { - // // log.Printf("RUNNING AGAIN!!") - // // runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) - // // return - // } - //} else { - // log.Printf("[WARNING] Failed getting cache for %s: %s", workflowExecution.ExecutionId, err) - //} - if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" { err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, true) //err = shuffle.SetWorkflowExecution(ctx, *workflowExecution, dbSave) @@ -1347,6 +1251,22 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request } } + sourceAuth, sourceAuthOk := request.URL.Query()["source_auth"] + if sourceAuthOk { + //log.Printf("\n\n\nSETTING SOURCE WORKFLOW AUTH TO %s!!!\n\n\n", sourceAuth[0]) + workflowExecution.ExecutionSourceAuth = sourceAuth[0] + } else { + //log.Printf("Did NOT get source workflow") + } + + sourceNode, sourceNodeOk := request.URL.Query()["source_node"] + if sourceNodeOk { + //log.Printf("\n\n\nSETTING SOURCE WORKFLOW NODE TO %s!!!\n\n\n", sourceNode[0]) + workflowExecution.ExecutionSourceNode = sourceNode[0] + } else { + //log.Printf("Did NOT get source workflow") + } + //workflowExecution.ExecutionSource = "default" sourceWorkflow, sourceWorkflowOk := request.URL.Query()["source_workflow"] if sourceWorkflowOk { @@ -1354,7 +1274,6 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request workflowExecution.ExecutionSource = sourceWorkflow[0] } else { //log.Printf("Did NOT get source workflow") - } sourceExecution, sourceExecutionOk := request.URL.Query()["source_execution"] @@ -1371,7 +1290,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request //log.Println(body) //if string(body)[0] == "\"" && string(body)[string(body) - log.Printf("[INFO] Body: %s", string(body)) + log.Printf("[DEBUG] Body: %s", string(body)) } var execution shuffle.ExecutionRequest @@ -2230,7 +2149,7 @@ func stopSchedule(resp http.ResponseWriter, request *http.Request) { if user.Id != workflow.Owner || len(user.Id) == 0 { if workflow.OrgId == user.ActiveOrg.Id && user.Role == "admin" { - log.Printf("[DEBUG] User %s is accessing workflow %s as admin (stop schedule)", user.Username, workflow.ID) + log.Printf("[AUDIT] User %s is accessing workflow %s as admin (stop schedule)", user.Username, workflow.ID) } else { log.Printf("[WARNING] Wrong user (%s) for workflow %s (stop schedule)", user.Username, workflow.ID) resp.WriteHeader(401) diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index a988ab6b..a653f569 100644 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -4880,6 +4880,7 @@ const AngularWorkflow = (props) => { if (actionlist.length === 0) { // FIXME: Have previous execution values in here actionlist.push({"type": "Execution Argument", "name": "Execution Argument", "value": "$exec", "highlight": "exec", "autocomplete": "exec", "example": "hello"}) + //actionlist.push({"type": "Key:Value store", "name": "Shuffle KV store", "value": "$shuffle_cache", "highlight": "shuffle_cache", "autocomplete": "shuffle_cache", "example": ""}) if (workflow.workflow_variables !== null && workflow.workflow_variables !== undefined && workflow.workflow_variables.length > 0) { for (var key in workflow.workflow_variables) { diff --git a/frontend/src/views/Workflows.jsx b/frontend/src/views/Workflows.jsx index c7b2dcb0..26a0baf7 100644 --- a/frontend/src/views/Workflows.jsx +++ b/frontend/src/views/Workflows.jsx @@ -359,6 +359,10 @@ const Workflows = (props) => { const [newWorkflowName, setNewWorkflowName] = React.useState(""); const [newWorkflowDescription, setNewWorkflowDescription] = React.useState(""); const [newWorkflowTags, setNewWorkflowTags] = React.useState([]); + + const [showExtraOptions, setShowExtraOptions] = React.useState(true); + const [defaultReturnValue, setDefaultReturnValue] = React.useState(""); + const [update, setUpdate] = React.useState("test"); const [deleteModalOpen, setDeleteModalOpen] = React.useState(false); const [publishModalOpen, setPublishModalOpen] = React.useState(false); @@ -368,6 +372,8 @@ const Workflows = (props) => { const [isDropzone, setIsDropzone] = React.useState(false); const [view, setView] = React.useState("grid") const [filters, setFilters] = React.useState([]) + const [submitLoading, setSubmitLoading] = React.useState(false) + const isCloud = window.location.host === "localhost:3002" || window.location.host === "shuffler.io" const findWorkflow = (filters) => { @@ -570,14 +576,14 @@ const Workflows = (props) => { } // Initialize the workflow itself - const ret = setNewWorkflow(data.name, data.description, data.tags, {}, false) + const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, {}, false) .then((response) => { if (response !== undefined) { // SET THE FULL THING data.id = response.id // Actually create it - const ret = setNewWorkflow(data.name, data.description, data.tags, data, false) + const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, data, false) .then((response) => { if (response !== undefined) { alert.success(`Successfully imported ${data.name}`) @@ -1233,10 +1239,12 @@ const Workflows = (props) => { }} > { + //console.log("DATA:" ,data) setModalOpen(true) setEditingWorkflow(data) setNewWorkflowName(data.name) setNewWorkflowDescription(data.description) + setDefaultReturnValue(data.default_return_value) if (data.tags !== undefined && data.tags !== null) { setNewWorkflowTags(JSON.parse(JSON.stringify(data.tags))) } @@ -1489,7 +1497,7 @@ const Workflows = (props) => { } // Can create and set workflows - const setNewWorkflow = (name, description, tags, editingWorkflow, redirect) => { + const setNewWorkflow = (name, description, tags, defaultReturnValue, editingWorkflow, redirect) => { var method = "POST" var extraData = "" @@ -1511,6 +1519,12 @@ const Workflows = (props) => { if (tags !== undefined) { workflowdata["tags"] = tags } + + if (defaultReturnValue !== undefined) { + workflowdata["default_return_value"] = defaultReturnValue + //console.log("WORKFLOW: ", workflowdata) + } + //console.log(workflowdata) //return @@ -1521,26 +1535,31 @@ const Workflows = (props) => { 'Accept': 'application/json', }, body: JSON.stringify(workflowdata), - credentials: "include", - }) + credentials: "include", + }) .then((response) => { if (response.status !== 200) { console.log("Status not 200 for workflows :O!") return } + setSubmitLoading(false) + return response.json() }) .then((responseJson) => { if (method === "POST" && redirect) { window.location.pathname = "/workflows/"+responseJson["id"] + setModalOpen(false) } else if (!redirect) { // Update :) setTimeout(() => { getAvailableWorkflows() }, 1000) setImportLoading(false) + setModalOpen(false) } else { alert.info("Successfully changed basic info for workflow") + setModalOpen(false) } return responseJson @@ -1548,6 +1567,8 @@ const Workflows = (props) => { .catch(error => { alert.error(error.toString()) setImportLoading(false) + setModalOpen(false) + setSubmitLoading(false) }); } @@ -1581,7 +1602,7 @@ const Workflows = (props) => { } // Initialize the workflow itself - const ret = setNewWorkflow(data.name, data.description, data.tags, {}, false) + const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, {}, false) .then((response) => { if (response !== undefined) { // SET THE FULL THING @@ -1591,7 +1612,7 @@ const Workflows = (props) => { data.is_valid = false // Actually create it - const ret = setNewWorkflow(data.name, data.description, data.tags, data, false) + const ret = setNewWorkflow(data.name, data.description, data.tags, data.default_return_value, data, false) .then((response) => { if (response !== undefined) { alert.success("Successfully imported "+data.name) @@ -1776,7 +1797,7 @@ const Workflows = (props) => { color="primary" defaultValue={newWorkflowDescription} placeholder="Description" - rows="6" + rows="3" multiline margin="dense" fullWidth @@ -1802,11 +1823,29 @@ const Workflows = (props) => { setUpdate("delete "+chip) }} /> + {showExtraOptions ? + setDefaultReturnValue(event.target.value)} + InputProps={{ + style:{ + color: "white", + }, + }} + color="primary" + defaultValue={defaultReturnValue} + placeholder="Default return value (used for Subflows if the subflow fails)" + rows="3" + multiline + margin="dense" + fullWidth + /> + : null} @@ -2065,7 +2110,7 @@ const Workflows = (props) => { return response.json() }) .then((responseJson) => { - console.log("DATA: ", responseJson) + //console.log("DATA: ", responseJson) if (!responseJson.success) { if (responseJson.reason !== undefined) { alert.error("Failed loading: "+responseJson.reason) @@ -2192,6 +2237,8 @@ const Workflows = (props) => { diff --git a/functions/onprem/worker/build.sh b/functions/onprem/worker/build.sh index 19a1e43d..cd38f58a 100644 --- a/functions/onprem/worker/build.sh +++ b/functions/onprem/worker/build.sh @@ -1,5 +1,5 @@ NAME=shuffle-worker -VERSION=0.9.10 +VERSION=0.9.12 echo "Running docker build with $NAME:$VERSION" #CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o worker.bin . diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index eb62aca7..468e5f51 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -12,9 +12,10 @@ require ( github.com/docker/docker v20.10.5+incompatible github.com/docker/go-connections v0.4.0 // indirect github.com/docker/go-units v0.4.0 // indirect - github.com/frikky/shuffle-shared v0.0.63 + github.com/frikky/shuffle-shared v0.0.86 github.com/fsouza/go-dockerclient v1.7.2 github.com/go-git/go-billy/v5 v5.3.1 // indirect + github.com/go-git/go-git/v5 v5.4.2 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/google/go-github/v28 v28.1.1 // indirect github.com/gorilla/mux v1.8.0 @@ -23,4 +24,5 @@ require ( github.com/patrickmn/go-cache v2.1.0+incompatible github.com/pkg/errors v0.9.1 // indirect google.golang.org/grpc v1.37.1 // indirect + gopkg.in/src-d/go-git.v4 v4.13.1 // indirect ) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 39b11a5c..dfbe8312 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -135,62 +135,67 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason log.Printf("[INFO] NOT cleaning up containers. IDS: %d, CLEANUP env: %s", len(containerIds), cleanupEnv) } - abortUrl := fmt.Sprintf("%s/api/v1/workflows/%s/executions/%s/abort", baseUrl, workflowExecution.Workflow.ID, workflowExecution.ExecutionId) + if len(reason) > 0 && len(nodeId) > 0 { + log.Printf("[INFO] Running abort of workflow because it should be finished") - path := fmt.Sprintf("?reason=%s", url.QueryEscape(reason)) - if len(nodeId) > 0 { - path += fmt.Sprintf("&node=%s", url.QueryEscape(nodeId)) - } - if len(environment) > 0 { - path += fmt.Sprintf("&env=%s", url.QueryEscape(environment)) - } - - //fmt.Println(url.QueryEscape(query)) - abortUrl += path - log.Printf("[INFO] Abort URL: %s", abortUrl) - - req, err := http.NewRequest( - "GET", - abortUrl, - nil, - ) - - if err != nil { - log.Println("[INFO] Failed building request: %s", err) - } - - // FIXME: Add an API call to the backend - authorization := os.Getenv("AUTHORIZATION") - if len(authorization) > 0 { - req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization)) - } else { - log.Printf("[ERROR] No authorization specified for abort") - } - - req.Header.Add("Content-Type", "application/json") - client := &http.Client{ - Transport: &http.Transport{ - Proxy: nil, - }, - } - - httpProxy := os.Getenv("HTTP_PROXY") - httpsProxy := os.Getenv("HTTPS_PROXY") - if (len(httpProxy) > 0 || len(httpsProxy) > 0) && baseUrl != "http://shuffle-backend:5001" { - client = &http.Client{} - } else { - if len(httpProxy) > 0 { - log.Printf("[INFO] Running with HTTP proxy %s (env: HTTP_PROXY)", httpProxy) + abortUrl := fmt.Sprintf("%s/api/v1/workflows/%s/executions/%s/abort", baseUrl, workflowExecution.Workflow.ID, workflowExecution.ExecutionId) + path := fmt.Sprintf("?reason=%s", url.QueryEscape(reason)) + if len(nodeId) > 0 { + path += fmt.Sprintf("&node=%s", url.QueryEscape(nodeId)) } - if len(httpsProxy) > 0 { - log.Printf("[INFO] Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy) + if len(environment) > 0 { + path += fmt.Sprintf("&env=%s", url.QueryEscape(environment)) } - } - log.Printf("[INFO] All App Logs: %#v", allLogs) - _, err = client.Do(req) - if err != nil { - log.Printf("[WARNING] Failed abort request: %s", err) + //fmt.Println(url.QueryEscape(query)) + abortUrl += path + log.Printf("[INFO] Abort URL: %s", abortUrl) + + req, err := http.NewRequest( + "GET", + abortUrl, + nil, + ) + + if err != nil { + log.Println("[INFO] Failed building request: %s", err) + } + + // FIXME: Add an API call to the backend + authorization := os.Getenv("AUTHORIZATION") + if len(authorization) > 0 { + req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", authorization)) + } else { + log.Printf("[ERROR] No authorization specified for abort") + } + + req.Header.Add("Content-Type", "application/json") + client := &http.Client{ + Transport: &http.Transport{ + Proxy: nil, + }, + } + + httpProxy := os.Getenv("HTTP_PROXY") + httpsProxy := os.Getenv("HTTPS_PROXY") + if (len(httpProxy) > 0 || len(httpsProxy) > 0) && baseUrl != "http://shuffle-backend:5001" { + client = &http.Client{} + } else { + if len(httpProxy) > 0 { + log.Printf("[INFO] Running with HTTP proxy %s (env: HTTP_PROXY)", httpProxy) + } + if len(httpsProxy) > 0 { + log.Printf("[INFO] Running with HTTPS proxy %s (env: HTTPS_PROXY)", httpsProxy) + } + } + + log.Printf("[INFO] All App Logs: %#v", allLogs) + _, err = client.Do(req) + if err != nil { + log.Printf("[WARNING] Failed abort request: %s", err) + } + } else { + log.Printf("[INFO] NOT running abort during shutdown.") } log.Printf("[INFO] Finished shutdown (after %d seconds). ", sleepDuration) @@ -761,6 +766,16 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { Value: workflowExecution.ExecutionId, }) + action.Parameters = append(action.Parameters, shuffle.WorkflowAppActionParameter{ + Name: "source_node", + Value: trigger.ID, + }) + + action.Parameters = append(action.Parameters, shuffle.WorkflowAppActionParameter{ + Name: "source_auth", + Value: workflowExecution.Authorization, + }) + //trigger.LargeImage = "" //err = handleSubworkflowExecution(client, workflowExecution, trigger, action) //if err != nil {