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

Added a way to run worker.go standalone: go run worker.go standalone <execution_id> <authentication> <url>

Browse Source
This commit is contained in:
Frikky
2025-04-05 00:08:03 +02:00
parent 557c39c8fc
commit 04c28b08c7
9 changed files with 363 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/go-app/go.mod
+1 -1
View File
@@ -6,7 +6,7 @@ toolchain go1.23.8
//replace github.com/frikky/schemaless => ../../../schemaless
//replace github.com/frikky/kin-openapi => ../../../../git/kin-openapi
//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
require (
cloud.google.com/go/datastore v1.20.0
backend/go-app/go.sum
-2
View File
@@ -341,8 +341,6 @@ github.com/sendgrid/sendgrid-go v3.14.0+incompatible h1:KDSasSTktAqMJCYClHVE94Fc
github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
github.com/shuffle/shuffle-shared v0.8.37 h1:RulNUpbBJAfYfArMMK/XUG1JE/vXTabcMA9oZ+Vxg0M=
github.com/shuffle/shuffle-shared v0.8.37/go.mod h1:NruHSAscDsW595wpK2r7MeHPGspUEKRNvBpcN1iGbHI=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
backend/go-app/main.go
+19 -10
View File
@@ -4241,12 +4241,17 @@ func runInitEs(ctx context.Context) {
continue
}
respBody, err := ioutil.ReadAll(newresp.Body)
if err != nil {
log.Printf("[ERROR] Failed setting respbody %s", err)
if newresp.StatusCode != 200 {
log.Printf("[WARNING] Failed stopping runs in environment %s. Status code: %d", environment, newresp.StatusCode)
continue
}
log.Printf("[DEBUG] Successfully ran workflow cleanup request for %s. Body: %s", environment, string(respBody))
//respBody, err := ioutil.ReadAll(newresp.Body)
//if err != nil {
// log.Printf("[ERROR] Failed setting respbody %s", err)
// continue
//}
//log.Printf("[DEBUG] Successfully ran workflow cleanup request for %s. Body: %s", environment, string(respBody))
url = fmt.Sprintf("http://localhost:%s/api/v1/environments/%s/rerun", backendPort, environment)
req, err = http.NewRequest(
@@ -4268,12 +4273,16 @@ func runInitEs(ctx context.Context) {
continue
}
respBody, err = ioutil.ReadAll(newresp.Body)
if err != nil {
log.Printf("[ERROR] Failed setting respbody %s", err)
continue
if newresp.StatusCode != 200 {
log.Printf("[WARNING] Failed rerunning environment %s. Status code: %d", environment, newresp.StatusCode)
}
//respBody, err := ioutil.ReadAll(newresp.Body)
//if err != nil {
// log.Printf("[ERROR] Failed setting respbody %s", err)
// continue
//}
//log.Printf("[DEBUG] Ran workflow RERUN request for %s with the response. Body: %s", environment, string(respBody))
}
}
@@ -5117,8 +5126,8 @@ func initHandlers() {
// Changed from workflows/streams to streams, as appengine was messing up
// This does not increase the API counter
// Used by frontend
r.HandleFunc("/api/v1/streams", handleWorkflowQueue).Methods("POST")
r.HandleFunc("/api/v1/streams/results", handleGetStreamResults).Methods("POST", "OPTIONS")
r.HandleFunc("/api/v1/streams", handleSetWorkflowExecution).Methods("POST")
r.HandleFunc("/api/v1/streams/results", handleGetWorkflowExecutionResult).Methods("POST", "OPTIONS")
// Used by orborus
r.HandleFunc("/api/v1/workflows/queue", handleGetWorkflowqueue).Methods("GET", "POST")
backend/go-app/walkoff.go
+15 -11
View File
@@ -548,7 +548,7 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) {
resp.Write(newjson)
}
func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
func handleGetWorkflowExecutionResult(resp http.ResponseWriter, request *http.Request) {
cors := shuffle.HandleCors(resp, request)
if cors {
return
@@ -679,7 +679,7 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) {
}
func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
func handleSetWorkflowExecution(resp http.ResponseWriter, request *http.Request) {
cors := shuffle.HandleCors(resp, request)
if cors {
return
@@ -698,20 +698,27 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
return
}
//log.Printf("Actionresult unmarshal: %s", string(body))
//log.Printf("[DEBUG] Got workflow result from %s of length %d", request.RemoteAddr, len(body))
// Allows override of existing executions.
// This is a way to set them back to 0 results and rerun the
// exact same. Primarily in use for Worker testing of specific workflows.
shouldReset := false
resetString, ok := request.URL.Query()["reset"]
if ok && len(resetString) > 0 {
if resetString[0] == "true" {
shouldReset = true
}
}
ctx := context.Background()
err = shuffle.ValidateNewWorkerExecution(ctx, body)
err = shuffle.ValidateNewWorkerExecution(ctx, body, shouldReset)
if err == nil {
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "success"}`)))
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Successfully updated the execution"}`)))
return
} else {
//log.Printf("[DEBUG] Handling other execution variant (subflow?): %s", err)
}
//log.Printf("[DEBUG] Got workflow result from %s of length %d.", request.RemoteAddr, len(body))
var actionResult shuffle.ActionResult
err = json.Unmarshal(body, &actionResult)
if err != nil {
@@ -721,8 +728,6 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
//return
}
//log.Printf("Received action: %#v", actionResult)
// 1. Get the WorkflowExecution(ExecutionId) from the database
// 2. if ActionResult.Authentication != WorkflowExecution.Authentication -> exit
// 3. Add to and update actionResult in workflowExecution
@@ -3036,7 +3041,6 @@ func executeSingleAction(resp http.ResponseWriter, request *http.Request) {
shouldRerun = true
}
workflowExecution, err := shuffle.PrepareSingleAction(ctx, user, fileId, body, runValidationAction)
debugUrl := fmt.Sprintf("/workflows/%s?execution_id=%s", workflowExecution.Workflow.ID, workflowExecution.ExecutionId)