#77: Added execution image to base
This commit is contained in:
+18
-7
@@ -3074,26 +3074,28 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
log.Printf("HOOK FOUND: %#v", hook)
|
//log.Printf("HOOK FOUND: %#v", hook)
|
||||||
// Execute the workflow
|
// Execute the workflow
|
||||||
//executeWorkflow(resp, request)
|
//executeWorkflow(resp, request)
|
||||||
|
|
||||||
//resp.WriteHeader(200)
|
//resp.WriteHeader(200)
|
||||||
//resp.Write([]byte(`{"success": true}`))
|
//resp.Write([]byte(`{"success": true}`))
|
||||||
if hook.Status == "stopped" {
|
if hook.Status == "stopped" {
|
||||||
|
log.Printf("Not running because hook status is stopped")
|
||||||
resp.WriteHeader(401)
|
resp.WriteHeader(401)
|
||||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "The webhook isn't running. Click start to start it"}`)))
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "The webhook isn't running. Click start to start it"}`)))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(hook.Workflows) == 0 {
|
if len(hook.Workflows) == 0 {
|
||||||
|
log.Printf("Not running because hook isn't connected to any workflows")
|
||||||
resp.WriteHeader(401)
|
resp.WriteHeader(401)
|
||||||
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No workflows are defined"}`)))
|
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No workflows are defined"}`)))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, item := range hook.Workflows {
|
for _, item := range hook.Workflows {
|
||||||
log.Printf("Running for workflow: %s", item)
|
log.Printf("Running for workflow %s with startnode %s", item, hook.Start)
|
||||||
workflow := Workflow{
|
workflow := Workflow{
|
||||||
ID: "",
|
ID: "",
|
||||||
}
|
}
|
||||||
@@ -3106,17 +3108,26 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
bodyWrapper := fmt.Sprintf(`{"start": "%s", "execution_argument": "%s"}`, hook.Start, string(body))
|
parsedBody := string(body)
|
||||||
if len(hook.Start) == 0 {
|
parsedBody = strings.Replace(parsedBody, "\"", "\\\"", -1)
|
||||||
bodyWrapper = string(body)
|
if len(parsedBody) > 0 {
|
||||||
|
if string(parsedBody[0]) == `"` && string(parsedBody[len(parsedBody)-1]) == "\"" {
|
||||||
|
parsedBody = parsedBody[1 : len(parsedBody)-1]
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
request := &http.Request{
|
bodyWrapper := fmt.Sprintf(`{"start": "%s", "execution_source": "webhook", "execution_argument": "%s"}`, hook.Start, string(parsedBody))
|
||||||
|
if len(hook.Start) == 0 {
|
||||||
|
log.Printf("No start node for hook %s - running with workflow default.", hook.Id)
|
||||||
|
bodyWrapper = string(parsedBody)
|
||||||
|
}
|
||||||
|
|
||||||
|
newRequest := &http.Request{
|
||||||
Method: "POST",
|
Method: "POST",
|
||||||
Body: ioutil.NopCloser(strings.NewReader(bodyWrapper)),
|
Body: ioutil.NopCloser(strings.NewReader(bodyWrapper)),
|
||||||
}
|
}
|
||||||
|
|
||||||
workflowExecution, executionResp, err := handleExecution(item, workflow, request)
|
workflowExecution, executionResp, err := handleExecution(item, workflow, newRequest)
|
||||||
|
|
||||||
if err == nil {
|
if err == nil {
|
||||||
err = increaseStatisticsField(ctx, "total_webhooks_ran", workflowExecution.Workflow.ID, 1)
|
err = increaseStatisticsField(ctx, "total_webhooks_ran", workflowExecution.Workflow.ID, 1)
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ var shuffleTestPath = "./shuffle-test-258209-5a2e8d7e508a.json"
|
|||||||
type ExecutionRequest struct {
|
type ExecutionRequest struct {
|
||||||
ExecutionId string `json:"execution_id"`
|
ExecutionId string `json:"execution_id"`
|
||||||
ExecutionArgument string `json:"execution_argument"`
|
ExecutionArgument string `json:"execution_argument"`
|
||||||
|
ExecutionSource string `json:"execution_source"`
|
||||||
WorkflowId string `json:"workflow_id"`
|
WorkflowId string `json:"workflow_id"`
|
||||||
Environments []string `json:"environments"`
|
Environments []string `json:"environments"`
|
||||||
Authorization string `json:"authorization"`
|
Authorization string `json:"authorization"`
|
||||||
@@ -153,6 +154,7 @@ type WorkflowExecution struct {
|
|||||||
Start string `json:"start" datastore:"start"`
|
Start string `json:"start" datastore:"start"`
|
||||||
ExecutionArgument string `json:"execution_argument" datastore:"execution_argument"`
|
ExecutionArgument string `json:"execution_argument" datastore:"execution_argument"`
|
||||||
ExecutionId string `json:"execution_id" datastore:"execution_id"`
|
ExecutionId string `json:"execution_id" datastore:"execution_id"`
|
||||||
|
ExecutionSource string `json:"execution_source" datastore:"execution_source"`
|
||||||
WorkflowId string `json:"workflow_id" datastore:"workflow_id"`
|
WorkflowId string `json:"workflow_id" datastore:"workflow_id"`
|
||||||
LastNode string `json:"last_node" datastore:"last_node"`
|
LastNode string `json:"last_node" datastore:"last_node"`
|
||||||
Authorization string `json:"authorization" datastore:"authorization"`
|
Authorization string `json:"authorization" datastore:"authorization"`
|
||||||
@@ -449,9 +451,9 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode
|
|||||||
// FIXME:
|
// FIXME:
|
||||||
// This may run multiple places if multiple servers,
|
// This may run multiple places if multiple servers,
|
||||||
// but that's a future problem
|
// but that's a future problem
|
||||||
log.Printf("BODY: %s", string(body))
|
//log.Printf("BODY: %s", string(body))
|
||||||
parsedArgument := strings.Replace(string(body), "\"", "\\\"", -1)
|
parsedArgument := strings.Replace(string(body), "\"", "\\\"", -1)
|
||||||
bodyWrapper := fmt.Sprintf(`{"start": "%s", "execution_argument": "%s"}`, startNode, parsedArgument)
|
bodyWrapper := fmt.Sprintf(`{"start": "%s", "execution_source": "schedule", "execution_argument": "%s"}`, startNode, parsedArgument)
|
||||||
log.Printf("WRAPPER BODY: \n%s", bodyWrapper)
|
log.Printf("WRAPPER BODY: \n%s", bodyWrapper)
|
||||||
job := func() {
|
job := func() {
|
||||||
request := &http.Request{
|
request := &http.Request{
|
||||||
@@ -1983,7 +1985,7 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
|
|||||||
var execution ExecutionRequest
|
var execution ExecutionRequest
|
||||||
err = json.Unmarshal(body, &execution)
|
err = json.Unmarshal(body, &execution)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
//log.Printf("Failed execution POST unmarshaling - still continue: %s", err)
|
log.Printf("Failed execution POST unmarshaling - continuing anyway: %s", err)
|
||||||
//return WorkflowExecution{}, "", err
|
//return WorkflowExecution{}, "", err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1992,11 +1994,15 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
|
|||||||
}
|
}
|
||||||
|
|
||||||
// FIXME - this should have "execution_argument" from executeWorkflow frontend
|
// FIXME - this should have "execution_argument" from executeWorkflow frontend
|
||||||
log.Printf("EXEC: %#v", execution)
|
//log.Printf("EXEC: %#v", execution)
|
||||||
if len(execution.ExecutionArgument) > 0 {
|
if len(execution.ExecutionArgument) > 0 {
|
||||||
workflowExecution.ExecutionArgument = execution.ExecutionArgument
|
workflowExecution.ExecutionArgument = execution.ExecutionArgument
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if len(execution.ExecutionSource) > 0 {
|
||||||
|
workflowExecution.ExecutionSource = execution.ExecutionSource
|
||||||
|
}
|
||||||
|
|
||||||
//log.Printf("Execution data: %#v", execution)
|
//log.Printf("Execution data: %#v", execution)
|
||||||
if len(execution.Start) == 36 {
|
if len(execution.Start) == 36 {
|
||||||
log.Printf("SHOULD START ON NODE %s", execution.Start)
|
log.Printf("SHOULD START ON NODE %s", execution.Start)
|
||||||
@@ -2144,6 +2150,13 @@ func handleExecution(id string, workflow Workflow, request *http.Request) (Workf
|
|||||||
workflowExecution.Status = "EXECUTING"
|
workflowExecution.Status = "EXECUTING"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if len(workflowExecution.ExecutionSource) == 0 {
|
||||||
|
log.Printf("No execution source specified. Setting to default")
|
||||||
|
workflowExecution.ExecutionSource = "default"
|
||||||
|
} else {
|
||||||
|
log.Printf("Execution source is %s for execution ID %s", workflowExecution.ExecutionSource, workflowExecution.ExecutionId)
|
||||||
|
}
|
||||||
|
|
||||||
workflowExecution.ExecutionVariables = workflow.ExecutionVariables
|
workflowExecution.ExecutionVariables = workflow.ExecutionVariables
|
||||||
// Local authorization for this single workflow used in workers.
|
// Local authorization for this single workflow used in workers.
|
||||||
|
|
||||||
|
|||||||
+113
-27
File diff suppressed because one or more lines are too long
@@ -24,6 +24,7 @@ const data = [{
|
|||||||
'curve-style': 'unbundled-bezier',
|
'curve-style': 'unbundled-bezier',
|
||||||
'label': 'data(label)',
|
'label': 'data(label)',
|
||||||
'text-margin-y': '-15px',
|
'text-margin-y': '-15px',
|
||||||
|
"color": "white",
|
||||||
"line-fill": "linear-gradient",
|
"line-fill": "linear-gradient",
|
||||||
"line-gradient-stop-colors": ["cyan", "yellow"],
|
"line-gradient-stop-colors": ["cyan", "yellow"],
|
||||||
"line-gradient-stop-positions": ["0.0", "100"],
|
"line-gradient-stop-positions": ["0.0", "100"],
|
||||||
|
|||||||
Reference in New Issue
Block a user