diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 0aca44b5..0466fb85 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -18187,8 +18187,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) { diff --git a/functions/kubernetes/charts/shuffle/README.md b/functions/kubernetes/charts/shuffle/README.md index 2e52d2f1..6c3a4ed2 100644 --- a/functions/kubernetes/charts/shuffle/README.md +++ b/functions/kubernetes/charts/shuffle/README.md @@ -383,7 +383,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` | diff --git a/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml b/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml index 244e037f..fd73dd15 100644 --- a/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml +++ b/functions/kubernetes/charts/shuffle/templates/orborus/orborus-dpl.yaml @@ -128,8 +128,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 }} @@ -138,25 +136,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 3e365794..661eefc6 100644 --- a/functions/kubernetes/charts/shuffle/values.schema.json +++ b/functions/kubernetes/charts/shuffle/values.schema.json @@ -1491,16 +1491,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 3d41d674..08763cb6 100644 --- a/functions/kubernetes/charts/shuffle/values.yaml +++ b/functions/kubernetes/charts/shuffle/values.yaml @@ -974,10 +974,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: diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 93baec04..2fbf51e3 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -2375,6 +2375,54 @@ 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() + 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 + } + + 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 { @@ -2419,10 +2467,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 { @@ -2468,10 +2520,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) @@ -2522,10 +2578,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) @@ -2843,10 +2903,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