From b46d3fe4ddffa1dd6605255ed98d13f6e4d28f93 Mon Sep 17 00:00:00 2001 From: Pascal Sthamer <10992664+P4sca1@users.noreply.github.com> Date: Fri, 11 Apr 2025 10:44:28 +0200 Subject: [PATCH 01/11] fix(helm): remove orborus container ports Signed-off-by: Pascal Sthamer <10992664+P4sca1@users.noreply.github.com> --- functions/kubernetes/charts/shuffle/README.md | 2 +- .../shuffle/templates/orborus/orborus-dpl.yaml | 11 ----------- .../templates/orborus/orborus-network-policy.yaml | 12 ------------ .../shuffle-worker-network-policy.yaml | 10 ---------- .../kubernetes/charts/shuffle/values.schema.json | 10 ---------- functions/kubernetes/charts/shuffle/values.yaml | 4 ---- 6 files changed, 1 insertion(+), 48 deletions(-) diff --git a/functions/kubernetes/charts/shuffle/README.md b/functions/kubernetes/charts/shuffle/README.md index ff3885d1..ecc8c265 100644 --- a/functions/kubernetes/charts/shuffle/README.md +++ b/functions/kubernetes/charts/shuffle/README.md @@ -381,7 +381,6 @@ The password should be provided with the `SHUFFLE_OPENSEARCH_PASSWORD` env varia | `orborus.image.pullPolicy` | orborus image pull policy | `IfNotPresent` | | `orborus.image.pullSecrets` | orborus image pull secrets | `[]` | | `orborus.replicaCount` | Number of orborus replicas to deploy | `1` | -| `orborus.containerPorts.http` | orborus HTTP container port | `8080` | | `orborus.extraContainerPorts` | Optionally specify extra list of additional ports for orborus containers | `[]` | | `orborus.livenessProbe.enabled` | Enable livenessProbe on orborus containers | `false` | | `orborus.livenessProbe.initialDelaySeconds` | Initial delay seconds for livenessProbe | `0` | @@ -607,3 +606,4 @@ The password should be provided with the `SHUFFLE_OPENSEARCH_PASSWORD` env varia ### Other Parameters + diff --git a/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml b/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml index a2d9c278..62a23242 100644 --- a/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml +++ b/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml @@ -110,8 +110,6 @@ spec: resources: {{- include "common.resources.preset" (dict "type" .Values.orborus.resourcesPreset) | nindent 12 }} {{- end }} ports: - - name: http - containerPort: {{ .Values.orborus.containerPorts.http }} {{- if .Values.orborus.extraContainerPorts }} {{- include "common.tplvalues.render" (dict "value" .Values.orborus.extraContainerPorts "context" $) | nindent 12 }} {{- end }} @@ -120,25 +118,16 @@ spec: livenessProbe: {{- include "common.tplvalues.render" (dict "value" .Values.orborus.customLivenessProbe "context" $) | nindent 12 }} {{- else if .Values.orborus.livenessProbe.enabled }} livenessProbe: {{- include "common.tplvalues.render" (dict "value" (omit .Values.orborus.livenessProbe "enabled") "context" $) | nindent 12 }} - httpGet: - path: / - port: {{ .Values.orborus.containerPorts.http }} {{- end }} {{- if .Values.orborus.customReadinessProbe }} readinessProbe: {{- include "common.tplvalues.render" (dict "value" .Values.orborus.customReadinessProbe "context" $) | nindent 12 }} {{- else if .Values.orborus.readinessProbe.enabled }} readinessProbe: {{- include "common.tplvalues.render" (dict "value" (omit .Values.orborus.readinessProbe "enabled") "context" $) | nindent 12 }} - httpGet: - path: / - port: {{ .Values.orborus.containerPorts.http }} {{- end }} {{- if .Values.orborus.customStartupProbe }} startupProbe: {{- include "common.tplvalues.render" (dict "value" .Values.orborus.customStartupProbe "context" $) | nindent 12 }} {{- else if .Values.orborus.startupProbe.enabled }} startupProbe: {{- include "common.tplvalues.render" (dict "value" (omit .Values.orborus.startupProbe "enabled") "context" $) | nindent 12 }} - httpGet: - path: / - port: {{ .Values.orborus.containerPorts.http }} {{- end }} {{- end }} {{- if .Values.orborus.lifecycleHooks }} diff --git a/functions/kubernetes/charts/shuffle/templates/orborus/orborus-network-policy.yaml b/functions/kubernetes/charts/shuffle/templates/orborus/orborus-network-policy.yaml index f6a22339..f55bf957 100644 --- a/functions/kubernetes/charts/shuffle/templates/orborus/orborus-network-policy.yaml +++ b/functions/kubernetes/charts/shuffle/templates/orborus/orborus-network-policy.yaml @@ -54,18 +54,6 @@ spec: {{- end }} {{- end }} ingress: - - ports: - - port: {{ .Values.orborus.containerPorts.http }} - protocol: TCP - {{- if not .Values.orborus.networkPolicy.allowExternal }} - from: - # Allow traffic from workers - - namespaceSelector: - matchLabels: - kubernetes.io/metadata.name: {{ .Release.Namespace }} - podSelector: - matchLabels: {{ include "shuffle.worker.matchLabels" . | nindent 14 }} - {{- end }} {{- if .Values.orborus.networkPolicy.extraIngress }} {{- include "common.tplvalues.render" ( dict "value" .Values.orborus.networkPolicy.extraIngress "context" $ ) | nindent 4 }} {{- end }} diff --git a/functions/kubernetes/charts/shuffle/templates/shuffle-worker/shuffle-worker-network-policy.yaml b/functions/kubernetes/charts/shuffle/templates/shuffle-worker/shuffle-worker-network-policy.yaml index 0ad96ada..3637d975 100644 --- a/functions/kubernetes/charts/shuffle/templates/shuffle-worker/shuffle-worker-network-policy.yaml +++ b/functions/kubernetes/charts/shuffle/templates/shuffle-worker/shuffle-worker-network-policy.yaml @@ -29,16 +29,6 @@ spec: - namespaceSelector: matchLabels: kubernetes.io/metadata.name: kube-system - # Allow access to orborus - - ports: - - port: {{ .Values.orborus.containerPorts.http }} - protocol: TCP - to: - - namespaceSelector: - matchLabels: - kubernetes.io/metadata.name: {{ .Release.Namespace }} - podSelector: - matchLabels: {{ include "shuffle.orborus.matchLabels" . | nindent 14 }} # Allow arbitrary connections to apps. Apps will typically use port 80/TCP, but this is not enforced. - to: - namespaceSelector: diff --git a/functions/kubernetes/charts/shuffle/values.schema.json b/functions/kubernetes/charts/shuffle/values.schema.json index b785b76e..c4687857 100644 --- a/functions/kubernetes/charts/shuffle/values.schema.json +++ b/functions/kubernetes/charts/shuffle/values.schema.json @@ -1471,16 +1471,6 @@ "description": "Number of orborus replicas to deploy", "default": 1 }, - "containerPorts": { - "type": "object", - "properties": { - "http": { - "type": "number", - "description": "orborus HTTP container port", - "default": 8080 - } - } - }, "extraContainerPorts": { "type": "array", "description": "Optionally specify extra list of additional ports for orborus containers", diff --git a/functions/kubernetes/charts/shuffle/values.yaml b/functions/kubernetes/charts/shuffle/values.yaml index 507c2468..7b2feaa5 100644 --- a/functions/kubernetes/charts/shuffle/values.yaml +++ b/functions/kubernetes/charts/shuffle/values.yaml @@ -958,10 +958,6 @@ orborus: ## @param orborus.replicaCount Number of orborus replicas to deploy ## replicaCount: 1 - ## @param orborus.containerPorts.http orborus HTTP container port - ## - containerPorts: - http: 8080 ## @param orborus.extraContainerPorts Optionally specify extra list of additional ports for orborus containers ## e.g: ## extraContainerPorts: From 4a180451265131d74651ccc9417defa3d1834ee9 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Wed, 16 Apr 2025 19:08:28 +0530 Subject: [PATCH 02/11] more info on app resp in the worker --- functions/onprem/worker/worker.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index a8f1c204..80aa5e56 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -3519,7 +3519,7 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, log.Printf("[ERROR] Failed reading app request body body: %s", err) return err } else { - log.Printf("[DEBUG][%s] NEWRESP (from app): %s", workflowExecution.ExecutionId, string(body)) + log.Printf("[DEBUG][%s] NEWRESP (from app %s with label %s): %s", workflowExecution.ExecutionId, action.AppName, action.Label, string(body)) } return nil From b508f8f81a84af1c2e839bdf254ba2288ac7df0c Mon Sep 17 00:00:00 2001 From: "lalitdeore12@gmail.com" Date: Fri, 2 May 2025 19:05:38 +0530 Subject: [PATCH 03/11] Fix - workflow page crashing issue for onprem --- frontend/src/views/AngularWorkflow.jsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 64d594bf..d1bd8c51 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -17780,8 +17780,8 @@ const AngularWorkflow = (defaultprops) => { - const defaultEnvironment = environments.find( - (env) => env.default && env.Name.toLowerCase() !== "cloud" + const defaultEnvironment = environments && environments?.find( + (env) => env?.default && env?.Name?.toLowerCase() !== "cloud" ); if (selectedTrigger.trigger_type === "PIPELINE" && selectedTrigger.environment === "onprem" && defaultEnvironment !== undefined) { From 622e019563baedd9c21b9b3c4ee99877e68aa582 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Wed, 7 May 2025 18:12:20 +0530 Subject: [PATCH 04/11] moved to new version of shuffle-shared --- backend/go-app/go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index 8e686705..ebc19034 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -22,7 +22,7 @@ require ( github.com/gorilla/mux v1.8.1 github.com/h2non/filetype v1.1.3 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.8.39 + github.com/shuffle/shuffle-shared v0.8.46 golang.org/x/crypto v0.36.0 google.golang.org/api v0.228.0 google.golang.org/grpc v1.71.1 From 6df7961793e67dbc8479d10d7124cbc96fff47da Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Tue, 13 May 2025 13:58:33 +0530 Subject: [PATCH 05/11] updated to new shuffle-shared version --- functions/onprem/orborus/go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 1d5d8afc..7cc2783e 100644 --- a/functions/onprem/orborus/go.mod +++ b/functions/onprem/orborus/go.mod @@ -10,7 +10,7 @@ require ( github.com/docker/docker v28.0.4+incompatible github.com/docker/go-connections v0.5.0 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.8.38 + github.com/shuffle/shuffle-shared v0.8.51 k8s.io/api v0.32.3 k8s.io/apimachinery v0.32.3 ) From 610e40ccf7ae2fb7524e2dcc473ec5df3005cf39 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Thu, 15 May 2025 13:45:36 +0530 Subject: [PATCH 06/11] bump the shuffle-shared version --- functions/onprem/orborus/go.mod | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 7cc2783e..fd65af94 100644 --- a/functions/onprem/orborus/go.mod +++ b/functions/onprem/orborus/go.mod @@ -10,7 +10,7 @@ require ( github.com/docker/docker v28.0.4+incompatible github.com/docker/go-connections v0.5.0 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.8.51 + github.com/shuffle/shuffle-shared v0.8.54 k8s.io/api v0.32.3 k8s.io/apimachinery v0.32.3 ) @@ -107,7 +107,7 @@ require ( go4.org v0.0.0-20201209231011-d4a079459e60 // indirect golang.org/x/crypto v0.36.0 // indirect golang.org/x/mod v0.21.0 // indirect - golang.org/x/net v0.37.0 // indirect + golang.org/x/net v0.38.0 // indirect golang.org/x/oauth2 v0.28.0 // indirect golang.org/x/sync v0.12.0 // indirect golang.org/x/sys v0.31.0 // indirect From bab9bead660b93f2a727e265be8a5eedc716db49 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Thu, 15 May 2025 15:27:35 +0530 Subject: [PATCH 07/11] bump shuffle-shared version for worker --- functions/onprem/worker/go.mod | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index 5db8570d..165889e7 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -10,7 +10,7 @@ require ( github.com/docker/docker v28.0.4+incompatible github.com/gorilla/mux v1.8.1 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.8.38 + github.com/shuffle/shuffle-shared v0.8.54 k8s.io/api v0.32.3 k8s.io/apimachinery v0.32.3 k8s.io/client-go v0.32.3 @@ -109,7 +109,7 @@ require ( go4.org v0.0.0-20201209231011-d4a079459e60 // indirect golang.org/x/crypto v0.36.0 // indirect golang.org/x/mod v0.21.0 // indirect - golang.org/x/net v0.37.0 // indirect + golang.org/x/net v0.38.0 // indirect golang.org/x/oauth2 v0.28.0 // indirect golang.org/x/sync v0.12.0 // indirect golang.org/x/sys v0.31.0 // indirect From c4c6dad43ba29282534547d81992dfd1d7b29d4e Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Thu, 15 May 2025 23:41:12 +0530 Subject: [PATCH 08/11] bumped and commented out not defined function for now --- backend/go-app/go.mod | 6 +++--- backend/go-app/main.go | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index ebc19034..46a339f1 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -22,7 +22,7 @@ require ( github.com/gorilla/mux v1.8.1 github.com/h2non/filetype v1.1.3 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.8.46 + github.com/shuffle/shuffle-shared v0.8.56 golang.org/x/crypto v0.36.0 google.golang.org/api v0.228.0 google.golang.org/grpc v1.71.1 @@ -114,7 +114,7 @@ require ( github.com/pjbgf/sha1cd v0.3.2 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect - github.com/sashabaranov/go-openai v1.19.2 // indirect + github.com/sashabaranov/go-openai v1.40.0 // indirect github.com/sendgrid/rest v2.6.9+incompatible // indirect github.com/sendgrid/sendgrid-go v3.14.0+incompatible // indirect github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect @@ -136,7 +136,7 @@ require ( go.opentelemetry.io/otel/trace v1.35.0 // indirect go.opentelemetry.io/proto/otlp v1.5.0 // indirect go4.org v0.0.0-20201209231011-d4a079459e60 // indirect - golang.org/x/net v0.37.0 // indirect + golang.org/x/net v0.38.0 // indirect golang.org/x/oauth2 v0.28.0 // indirect golang.org/x/sync v0.12.0 // indirect golang.org/x/sys v0.31.0 // indirect diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 68fc4e88..f6f22f73 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -5348,7 +5348,7 @@ func initHandlers() { r.HandleFunc("/api/v1/users/notifications/clear", shuffle.HandleClearNotifications).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/users/notifications/{notificationId}/markasread", shuffle.HandleMarkAsRead).Methods("GET", "OPTIONS") - r.HandleFunc("/api/v1/conversation", shuffle.RunActionAI).Methods("POST", "OPTIONS") + //r.HandleFunc("/api/v1/conversation", shuffle.RunActionAI).Methods("POST", "OPTIONS") //r.HandleFunc("/api/v1/users/notifications/{notificationId}/markasread", shuffle.HandleMarkAsRead).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/dashboards/{key}/widgets", shuffle.HandleNewWidget).Methods("POST", "OPTIONS") From 514c018f703b7c557e900044fe0c6d3cd75d73e0 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Fri, 16 May 2025 12:20:38 +0530 Subject: [PATCH 09/11] fix: commented out the wrong endpoint --- backend/go-app/main.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/go-app/main.go b/backend/go-app/main.go index f6f22f73..fbd2a251 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -5138,7 +5138,7 @@ func initHandlers() { r.HandleFunc("/api/v1/apps/{key}/execute", executeSingleAction).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/apps/{key}/run", executeSingleAction).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/apps/categories", shuffle.GetActiveCategories).Methods("GET", "OPTIONS") - r.HandleFunc("/api/v1/apps/categories/run", shuffle.RunCategoryAction).Methods("POST", "OPTIONS") + //r.HandleFunc("/api/v1/apps/categories/run", shuffle.RunCategoryAction).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/apps/upload", handleAppZipUpload).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/apps/{appId}/activate", activateWorkflowAppDocker).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/apps/{appId}/deactivate", activateWorkflowAppDocker).Methods("GET", "OPTIONS") @@ -5348,7 +5348,7 @@ func initHandlers() { r.HandleFunc("/api/v1/users/notifications/clear", shuffle.HandleClearNotifications).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/users/notifications/{notificationId}/markasread", shuffle.HandleMarkAsRead).Methods("GET", "OPTIONS") - //r.HandleFunc("/api/v1/conversation", shuffle.RunActionAI).Methods("POST", "OPTIONS") + r.HandleFunc("/api/v1/conversation", shuffle.RunActionAI).Methods("POST", "OPTIONS") //r.HandleFunc("/api/v1/users/notifications/{notificationId}/markasread", shuffle.HandleMarkAsRead).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/dashboards/{key}/widgets", shuffle.HandleNewWidget).Methods("POST", "OPTIONS") From af6db1d43285e70231d038f69a4293f82b39bbc9 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Wed, 21 May 2025 22:14:07 +0530 Subject: [PATCH 10/11] fix: requesting backend if we cannot find the executionId on cache --- functions/onprem/worker/worker.go | 92 +++++++++++++++++++++++++------ 1 file changed, 76 insertions(+), 16 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 80aa5e56..6c2681eb 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -2336,6 +2336,50 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar { } return envVars } +func getWorkerBackendExecution(auth string, executionId string) (*shuffle.WorkflowExecution, error) { + backendUrl := os.Getenv("BASE_URL") + if len(backendUrl) == 0 { + backendUrl = "http://shuffle-backend:5001" + } + + var workflowExecution *shuffle.WorkflowExecution + + streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", backendUrl) + topClient := shuffle.GetExternalClient(backendUrl) + requestData := shuffle.ActionResult { + Authorization: auth, + ExecutionId: executionId, + } + + data, err := json.Marshal(requestData) + if err != nil { + return workflowExecution, err + } + + req, err := http.NewRequest( + "POST", + streamResultUrl, + bytes.NewBuffer([]byte(data)), + ) + + newresp, err := topClient.Do(req) + if err != nil { + return workflowExecution, err + } + + defer newresp.Body.Close() + body, err := ioutil.ReadAll(newresp.Body) + if err != nil { + return workflowExecution, err + } + + err = json.Unmarshal(body, &workflowExecution) + if err != nil { + return workflowExecution, err + } + + return workflowExecution, nil +} func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { if request.Body == nil { @@ -2380,10 +2424,14 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) { workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId) if err != nil { - log.Printf("[ERROR][%s] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, actionResult.ExecutionId, err) - resp.WriteHeader(500) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist locally."}`, actionResult.ExecutionId))) - return + log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (1): %s", actionResult.ExecutionId, err) + workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId) + if err != nil { + log.Printf("[ERROR][%s] Failed getting execution (workflowqueue) %s: %s", actionResult.ExecutionId, actionResult.ExecutionId, err) + resp.WriteHeader(500) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution ID %s because it doesn't exist locally."}`, actionResult.ExecutionId))) + return + } } if workflowExecution.Authorization != actionResult.Authorization { @@ -2429,10 +2477,14 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl //log.Printf("[DEBUG][%s] IN WORKFLOWEXECUTION SUB!", actionResult.ExecutionId) workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId) if err != nil { - log.Printf("[ERROR] Failed getting execution cache: %s", err) - resp.WriteHeader(400) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`))) - return + log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (2): %s", actionResult.ExecutionId, err) + workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId) + if err != nil { + log.Printf("[ERROR] Failed getting execution cache: %s", err) + resp.WriteHeader(400) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution"}`))) + return + } } resultLength := len(workflowExecution.Results) @@ -2483,10 +2535,14 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl if strings.Contains(fmt.Sprintf("%s", err), "Rerun this transaction") { workflowExecution, err := shuffle.GetWorkflowExecution(ctx, workflowExecutionId) if err != nil { - log.Printf("[ERROR][%s] Failed getting execution cache (2): %s", workflowExecution.ExecutionId, err) - resp.WriteHeader(400) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution (2)"}`))) - return + log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (3): %s", actionResult.ExecutionId, err) + workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId) + if err != nil { + log.Printf("[ERROR][%s] Failed getting execution cache (2): %s", workflowExecution.ExecutionId, err) + resp.WriteHeader(400) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Failed getting execution (2)"}`))) + return + } } resultLength = len(workflowExecution.Results) @@ -2796,10 +2852,14 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) { ctx := context.Background() workflowExecution, err := shuffle.GetWorkflowExecution(ctx, actionResult.ExecutionId) if err != nil { - log.Printf("[INFO] Failed getting execution (streamresult) %s: %s", actionResult.ExecutionId, err) - resp.WriteHeader(400) - resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`))) - return + log.Printf("[WARNING][%s] Failed to find execution in cache requesting backend (4): %s", actionResult.ExecutionId, err) + workflowExecution, err = getWorkerBackendExecution(actionResult.Authorization, actionResult.ExecutionId) + if err != nil { + log.Printf("[ERROR] Failed getting execution (streamresult) %s: %s", actionResult.ExecutionId, err) + resp.WriteHeader(400) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Bad authorization key or execution_id might not exist."}`))) + return + } } // Authorization is done here From 0344cc789b8e1d4427ed9a8efcdff7550a1b9f33 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Wed, 21 May 2025 22:22:17 +0530 Subject: [PATCH 11/11] handling bad status code --- functions/onprem/worker/worker.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 6c2681eb..cde2484c 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -2368,6 +2368,10 @@ func getWorkerBackendExecution(auth string, executionId string) (*shuffle.Workfl } defer newresp.Body.Close() + if newresp.StatusCode != 200 { + return workflowExecution, errors.New(fmt.Sprintf("Got bad status code from backend %d", newresp.StatusCode)) + } + body, err := ioutil.ReadAll(newresp.Body) if err != nil { return workflowExecution, err