marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed non finishing workflows with conditions

Browse Source
This commit is contained in:
Frikky
2024-01-17 14:27:22 +01:00
parent 14498f1fb4
commit d9b7123382
2 changed files with 15 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/onprem/worker/go.mod
+1 -1
View File
@@ -11,7 +11,7 @@ require (
github.com/gorilla/mux v1.8.0 github.com/gorilla/mux v1.8.0
github.com/patrickmn/go-cache v2.1.0+incompatible github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/satori/go.uuid v1.2.0 github.com/satori/go.uuid v1.2.0
github.com/shuffle/shuffle-shared v0.5.53 github.com/shuffle/shuffle-shared v0.5.57
k8s.io/api v0.28.3 k8s.io/api v0.28.3
k8s.io/apimachinery v0.28.3 k8s.io/apimachinery v0.28.3
k8s.io/client-go v0.28.3 k8s.io/client-go v0.28.3
functions/onprem/worker/worker.go
Executable → Regular
+14 -3
View File
@@ -1501,6 +1501,8 @@ func executionInit(workflowExecution shuffle.WorkflowExecution) error {
} }
func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.WorkflowExecution, streamResultUrl, subflowId string) error { func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.WorkflowExecution, streamResultUrl, subflowId string) error {
// FIXME: If MEMCACHE is enabled, check in this order:
extra := 0 extra := 0
for _, trigger := range workflowExecution.Workflow.Triggers { for _, trigger := range workflowExecution.Workflow.Triggers {
if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" { if trigger.AppName == "User Input" || trigger.AppName == "Shuffle Workflow" {
@@ -2064,7 +2066,8 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl
} }
if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" { if setExecution || workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE" {
log.Printf("[DEBUG][%s] Running setexec with status %s and %d result(s)", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results)) log.Printf("[DEBUG][%s] Running setexec with status %s and %d/%d results", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions))
//result(s)", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results))
err = setWorkflowExecution(ctx, *workflowExecution, dbSave) err = setWorkflowExecution(ctx, *workflowExecution, dbSave)
if err != nil { if err != nil {
resp.WriteHeader(401) resp.WriteHeader(401)
@@ -2360,6 +2363,8 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
port := listener.Addr().(*net.TCPAddr).Port port := listener.Addr().(*net.TCPAddr).Port
// Set the port environment variable
os.Setenv("WORKER_PORT", fmt.Sprintf("%d", port))
log.Printf("[DEBUG] Starting webserver (2) on port %d with hostname: %s", port, hostname) log.Printf("[DEBUG] Starting webserver (2) on port %d with hostname: %s", port, hostname)
appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, port) appCallbackUrl = fmt.Sprintf("http://%s:%d", hostname, port)
@@ -2605,7 +2610,7 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int,
client := shuffle.GetExternalClient(streamUrl) client := shuffle.GetExternalClient(streamUrl)
// Set client timeout to 5 seconds // Set client timeout to 5 seconds
client.Timeout = time.Duration(10) * time.Second //client.Timeout = time.Duration(10) * time.Second
newresp, err := client.Do(req) newresp, err := client.Do(req)
if err != nil { if err != nil {
// Another timeout issue here somewhere // Another timeout issue here somewhere
@@ -2807,6 +2812,7 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe
log.Printf("[DEBUG] Environments: %s. Source: %s. 1 env = webserver, 0 or >1 = default. Subflow exists: %#v", environments, workflowExecution.ExecutionSource, subflowFound) log.Printf("[DEBUG] Environments: %s. Source: %s. 1 env = webserver, 0 or >1 = default. Subflow exists: %#v", environments, workflowExecution.ExecutionSource, subflowFound)
if len(environments) == 1 && workflowExecution.ExecutionSource != "default" && !subflowFound { if len(environments) == 1 && workflowExecution.ExecutionSource != "default" && !subflowFound {
log.Printf("[DEBUG] Running OPTIMIZED execution (not manual)") log.Printf("[DEBUG] Running OPTIMIZED execution (not manual)")
os.Setenv("SHUFFLE_OPTIMIZED", "true")
listener := webserverSetup(workflowExecution) listener := webserverSetup(workflowExecution)
err := executionInit(workflowExecution) err := executionInit(workflowExecution)
if err != nil { if err != nil {
@@ -2820,7 +2826,13 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe
handleExecutionResult(workflowExecution) handleExecutionResult(workflowExecution)
}() }()
log.Printf("[DEBUG] Running with port %#v", os.Getenv("WORKER_PORT"))
runWebserver(listener) runWebserver(listener)
// Set environment variable
//log.Printf("Before wait") //log.Printf("Before wait")
//wg := sync.WaitGroup{} //wg := sync.WaitGroup{}
//wg.Add(1) //wg.Add(1)
@@ -3262,5 +3274,4 @@ func runWebserver(listener net.Listener) {
if err != nil { if err != nil {
log.Printf("[ERROR] Serve issue in worker: %#v", err) log.Printf("[ERROR] Serve issue in worker: %#v", err)
} }
log.Printf("[DEBUG] Do we see this?")
} }