From 67cd5a3e60a946b4c9a39a19b5ae12b60468bbec Mon Sep 17 00:00:00 2001 From: Frikky Date: Mon, 1 Sep 2025 19:19:34 +0200 Subject: [PATCH] Synced all 2.1.0 fixes --- backend/go-app/go.mod | 2 +- backend/go-app/go.sum | 4 +- frontend/src/components/EnvironmentTab.jsx | 112 ++++++----- frontend/src/views/RunWorkflow.jsx | 204 ++++++++++++--------- functions/onprem/orborus/go.mod | 2 +- functions/onprem/orborus/go.sum | 4 +- functions/onprem/orborus/orborus.go | 1 - functions/onprem/worker/go.mod | 2 +- functions/onprem/worker/go.sum | 4 +- functions/onprem/worker/worker.go | 82 ++++++--- 10 files changed, 251 insertions(+), 166 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index d978245c..73fe8417 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -24,7 +24,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.9.14 + github.com/shuffle/shuffle-shared v0.9.15 github.com/shuffle/singul v0.0.16 golang.org/x/crypto v0.40.0 google.golang.org/api v0.236.0 diff --git a/backend/go-app/go.sum b/backend/go-app/go.sum index 71508c8f..871227bc 100644 --- a/backend/go-app/go.sum +++ b/backend/go-app/go.sum @@ -363,8 +363,8 @@ github.com/sendgrid/sendgrid-go v3.16.1+incompatible h1:zWhTmB0Y8XCDzeWIm2/BIt1G github.com/sendgrid/sendgrid-go v3.16.1+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.9.14 h1:POkTHO+bByuv8HiKuCMSGgtpDKk86ISr6ooLG8vQfuE= -github.com/shuffle/shuffle-shared v0.9.14/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw= +github.com/shuffle/shuffle-shared v0.9.15 h1:Gc7c0pbWG6nHWSTkcfAnKgQgWCWfc6aDQ/BIu20z6bM= +github.com/shuffle/shuffle-shared v0.9.15/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw= github.com/shuffle/singul v0.0.16 h1:dW+0Mln9R1aUJ0fjikpWcxbjoQWqJHxe4kSxh2tQN5E= github.com/shuffle/singul v0.0.16/go.mod h1:LYkp320A6gsoPlYbXUM+WvEPUVAuutlSsqnVKyRy4gs= github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= diff --git a/frontend/src/components/EnvironmentTab.jsx b/frontend/src/components/EnvironmentTab.jsx index 707512ce..c85763df 100644 --- a/frontend/src/components/EnvironmentTab.jsx +++ b/frontend/src/components/EnvironmentTab.jsx @@ -38,7 +38,7 @@ import { } from "@mui/icons-material"; import { toast } from 'react-toastify'; import { Context } from '../context/ContextApi.jsx'; -import { green, red } from '../views/AngularWorkflow.jsx' +import { green, yellow, red } from '../views/AngularWorkflow.jsx' import AppSearch from "../components/AppSearch1.jsx"; const EnvironmentTab = memo((props) => { @@ -612,13 +612,6 @@ const EnvironmentTab = memo((props) => { } }; - const queueSizeText = (queue) => { - if (queue === undefined || queue === null) return 0; - if (queue < 0) return 0; - if (queue > 1000) return ">1000"; - return queue; - }; - const LocationActionModal = (props) => { const { showLocationActionModal } = props @@ -990,11 +983,14 @@ const EnvironmentTab = memo((props) => { } const queueSize = + selectedOrganization.id !== undefined && environment?.org_id !== selectedOrganization.id ? + "N/A" + : environment.queue !== undefined && environment.queue !== null ? environment.queue < 0 ? 0 - : environment.queue > 1000 - ? ">1000" + : environment.queue >= 100 + ? ">100" : environment.queue : 0; @@ -1088,11 +1084,13 @@ const EnvironmentTab = memo((props) => { {environment.Type !== "cloud" - ? environment.running_ip === undefined || + ? + selectedOrganization.id !== undefined && environment?.org_id !== selectedOrganization.id ? "Parent Org Controlled. Status not available in suborgs." + : environment.running_ip === undefined || environment.running_ip === null || environment.running_ip.length === 0 ? - "Not running. Click to get the start command that can be ran on your server." + "Probably not running. Check your Orborus instance." : IP / label: {environment?.running_ip?.split(":")[0]}. May stay running up to a minute after stopping Orborus. : @@ -1102,7 +1100,7 @@ const EnvironmentTab = memo((props) => {

- Last checkin: {environment?.checkin !== undefined && environment.checkin !== null && environment?.checkin > 0 ? new Date(environment?.checkin * 1000).toLocaleString() : "Never"} + Last checkin: {environment?.checkin !== undefined && environment.checkin !== null && environment?.checkin > 0 ? new Date(environment?.checkin * 1000).toLocaleString() : "Never"} {environment?.Type === "cloud" ? "" : "Timeout: 180 seconds"} } placement="top"> { }} variant="body2" > - {environment.Type !== "cloud" && + {environment.Type === "cloud" ? + { + //handleChipClick + }} + variant="outlined" + color="primary" + /> + : + selectedOrganization.id !== undefined && environment?.org_id !== selectedOrganization.id ? + { + //handleChipClick + }} + variant="outlined" + color="primary" + /> + : (environment.running_ip === undefined || environment.running_ip === null || environment.running_ip.length === 0) @@ -1157,35 +1184,38 @@ const EnvironmentTab = memo((props) => { - - - ) : ( - - - - - - ) - } - style={{ - minWidth: 60, - marginLeft: 20, - overflow: "hidden", - whiteSpace: "normal", - wordWrap: "break-word", - padding: 8, - display: "table-cell", - }} - /> + + + ) : ( + + + + + + ) + } + style={{ + minWidth: 60, + marginLeft: 20, + overflow: "hidden", + whiteSpace: "normal", + wordWrap: "break-word", + padding: 8, + display: "table-cell", + }} + /> { const [boxWidth, setBoxWidth] = React.useState(500) const [inputQuestions, setInputQuestions] = React.useState([]) + const searchParams = new URLSearchParams(window.location.search) + const answer = searchParams.get("answer") + const execution_id = searchParams.get("reference_execution") + const authorization = searchParams.get("authorization") + const sourceNode = searchParams.get("source_node") + const backendUrl = searchParams.get("backend_url") || globalUrl useEffect(() => { if (workflow === undefined || workflow === null || Object.keys(workflow).length === 0) { @@ -173,7 +179,7 @@ const RunWorkflow = (defaultprops) => { try { executionArgument = JSON.parse(executionArgument) } catch (e) { - console.log("Error parsing execution argument: ", e) + //console.log("Error parsing execution argument: ", e) executionArgument = {} } } @@ -189,7 +195,7 @@ const RunWorkflow = (defaultprops) => { } const getWorkflows = () => { - const url = `${globalUrl}/api/v1/workflows` + const url = `${backendUrl}/api/v1/workflows` fetch(url, { method: "GET", headers: { @@ -207,7 +213,7 @@ const RunWorkflow = (defaultprops) => { }) .then((responseJson) => { if (responseJson.success === false) { - toast.error("Failed saving workflow. Please try again.") + //toast.error("Failed getting workflows. Please try again.") } else { if (responseJson?.length > 0) { setWorkflows(responseJson) @@ -220,7 +226,7 @@ const RunWorkflow = (defaultprops) => { } const loadForms = (orgId) => { - const url = `${globalUrl}/api/v1/orgs/${orgId}/forms` + const url = `${backendUrl}/api/v1/orgs/${orgId}/forms` fetch(url, { method: "GET", headers: { @@ -238,7 +244,7 @@ const RunWorkflow = (defaultprops) => { }) .then((responseJson) => { if (responseJson.success === false) { - toast.error("Failed saving workflow. Please try again.") + //toast.error("Failed loading forms. Please try again or contact support@shuffler.io if this persists.") } else { if (responseJson?.length > 0) { // Sort them by name @@ -253,7 +259,7 @@ const RunWorkflow = (defaultprops) => { } const saveWorkflow = (workflow) => { - const url = `${globalUrl}/api/v1/workflows/${workflow.id}` + const url = `${backendUrl}/api/v1/workflows/${workflow.id}` fetch(url, { method: "PUT", headers: { @@ -283,7 +289,7 @@ const RunWorkflow = (defaultprops) => { } const getApps = () => { - fetch(globalUrl + "/api/v1/apps", { + fetch(backendUrl+ "/api/v1/apps", { method: "GET", headers: { "Content-Type": "application/json", @@ -423,7 +429,7 @@ const RunWorkflow = (defaultprops) => { */ } - var url = `${globalUrl}/api/v1/workflows/${props.match.params.key}/run` + var url = `${backendUrl}/api/v1/workflows/${props.match.params.key}/run` var fetchBody = { headers: { 'Content-Type': 'application/json; charset=utf-8', @@ -474,8 +480,8 @@ const RunWorkflow = (defaultprops) => { } } - if (response.status === 401 || response.status === 403) { - toast(`This Form is not available for you to run. If you this is an error, contact ${supportEmail} with a link to this form`) + if ((response.status === 401 || response.status === 403) && authorization === undefined || authorization === null || authorization.length === 0) { + toast(`This form is not available for you to run. If you this is an error, contact ${supportEmail} with a link to this form`) } return response.json() @@ -489,11 +495,15 @@ const RunWorkflow = (defaultprops) => { if (responseJson.success === false) { console.log("Failed sending execution request") if (responseJson.reason !== undefined && responseJson.reason !== null) { - toast.warn(responseJson.reason) + if (responseJson?.reason?.toLowerCase().includes("already clicked")) { + setMessage("Already answered. You may close this window (2).") + } else { + toast.warn(responseJson.reason) + } } stop() - setMessage("") + //setMessage("") setExecutionData({}) setExecutionInfo("") setExecutionRunning(false) @@ -556,7 +566,7 @@ const RunWorkflow = (defaultprops) => { const loadInputWorkflowData = (workflow_id, inputWorkflow) => { - const url = `${globalUrl}/api/v1/workflows/${workflow_id}/run` + const url = `${backendUrl}/api/v1/workflows/${workflow_id}/run` fetch(url, { method: "POST", headers: { @@ -607,58 +617,13 @@ const RunWorkflow = (defaultprops) => { }) } - const searchParams = new URLSearchParams(window.location.search) - const answer = searchParams.get("answer") - const execution_id = searchParams.get("reference_execution") - const authorization = searchParams.get("authorization") - const sourceNode = searchParams.get("source_node") + const setupSourcenode = (workflow, selectedNode) => { - const getWorkflow = (workflow_id, selectedNode) => { - setRealtimeMarkdown("") - - const url = `${globalUrl}/api/v1/workflows/${workflow_id}` - fetch(url, { - method: "GET", - headers: { - "Content-Type": "application/json", - Accept: "application/json", - }, - credentials: "include", - }) - .then((response) => { - if (response.status !== 200) { - console.log("Status not 200 for workflows :O!"); - } - - if (response.status === 401 || response.status === 403) { - toast(`This Form is not available to you. If you think this is an error, please contact ${supportEmail} with the URL.`) - } - - return response.json() - }) - .then((responseJson) => { - // Not sure why this is necessary. - if (responseJson.isValid === undefined) { - responseJson.isValid = true; - } - - if (responseJson.errors === undefined) { - responseJson.errors = []; - } - - if (responseJson.actions === undefined || responseJson.actions === null) { - responseJson.actions = []; - } - - if (responseJson.triggers === undefined || responseJson.triggers === null) { - responseJson.triggers = []; - } - - if (responseJson.input_questions !== undefined && responseJson.input_questions !== null && responseJson.input_questions.length > 0) { + if (workflow.input_questions !== undefined && workflow.input_questions !== null && workflow.input_questions.length > 0) { var newexec = {} - for (let questionkey in responseJson.input_questions) { - const question = responseJson.input_questions[questionkey] + for (let questionkey in workflow.input_questions) { + const question = workflow.input_questions[questionkey] var multiChoiceOptions = question.value !== undefined && question.value !== null && question.value.length > 0 && question.value.includes(";") ? question.value.split(";") : [] if (multiChoiceOptions.length > 1) { @@ -670,8 +635,8 @@ const RunWorkflow = (defaultprops) => { // Override with just relevant fields if (sourceNode !== undefined && sourceNode !== null && sourceNode.length > 0) { - for (var triggerkey in responseJson.triggers) { - const trig = responseJson.triggers[triggerkey] + for (var triggerkey in workflow.triggers) { + const trig = workflow.triggers[triggerkey] if (trig.id !== sourceNode) { continue } @@ -693,8 +658,8 @@ const RunWorkflow = (defaultprops) => { const parsed = JSON.parse(param.value) // Find this in the workflow.input_questions - for (var questionkey in responseJson.input_questions) { - var question = JSON.parse(JSON.stringify(responseJson.input_questions[questionkey])) + for (var questionkey in workflow.input_questions) { + var question = JSON.parse(JSON.stringify(workflow.input_questions[questionkey])) question.value = question.value.split(";")[0] if (parsed.includes(question.name)) { keepfields.push(question.value) @@ -724,38 +689,38 @@ const RunWorkflow = (defaultprops) => { if (selectedNode !== undefined && selectedNode !== null && selectedNode.length > 0) { var found = false - for (var actionkey in responseJson.actions) { - if (responseJson.actions[actionkey].id === selectedNode) { + for (var actionkey in workflow.actions) { + if (workflow.actions[actionkey].id === selectedNode) { found = true - setFoundSourcenode(responseJson.actions[actionkey]) + setFoundSourcenode(workflow.actions[actionkey]) break } } if (!found) { - for (var triggerkey in responseJson.triggers) { - if (responseJson.triggers[triggerkey].id !== selectedNode) { + for (var triggerkey in workflow.triggers) { + if (workflow.triggers[triggerkey].id !== selectedNode) { continue } - setFoundSourcenode(responseJson.triggers[triggerkey]) + setFoundSourcenode(workflow.triggers[triggerkey]) - if (responseJson.input_questions !== undefined && responseJson.input_questions !== null && responseJson.input_questions.length > 0 && responseJson.triggers[triggerkey].trigger_type === "USERINPUT") { + if (workflow.input_questions !== undefined && workflow.input_questions !== null && workflow.input_questions.length > 0 && workflow.triggers[triggerkey].trigger_type === "USERINPUT") { // Look for input questions param - for (var paramkey in responseJson.triggers[triggerkey].parameters) { - if (responseJson.triggers[triggerkey].parameters[paramkey].name === "input_questions") { + for (var paramkey in workflow.triggers[triggerkey].parameters) { + if (workflow.triggers[triggerkey].parameters[paramkey].name === "input_questions") { var relevantquestions = [] - for (var questionkey in responseJson.input_questions) { - if (responseJson.triggers[triggerkey].parameters[paramkey].value.includes(responseJson.input_questions[questionkey].name)) { - relevantquestions.push(responseJson.input_questions[questionkey]) + for (var questionkey in workflow.input_questions) { + if (workflow.triggers[triggerkey].parameters[paramkey].value.includes(workflow.input_questions[questionkey].name)) { + relevantquestions.push(workflow.input_questions[questionkey]) } } setInputQuestions(relevantquestions) - //responseJson.input_questions = relevantquestions + //workflow.input_questions = relevantquestions } } } @@ -765,13 +730,13 @@ const RunWorkflow = (defaultprops) => { } } } else { - setInputQuestions(responseJson.input_questions) + setInputQuestions(workflow.input_questions) } - if (responseJson.form_control.input_markdown !== undefined && responseJson.form_control.input_markdown !== null && responseJson.form_control.input_markdown.length > 0) { + if (workflow.form_control.input_markdown !== undefined && workflow.form_control.input_markdown !== null && workflow.form_control.input_markdown.length > 0) { // Look for {{ uuid }} format, and try to run that workflow with their account // This is a hack, but a fun one. - var newmarkdown = responseJson.form_control.input_markdown.replace("", "") + var newmarkdown = workflow.form_control.input_markdown.replace("", "") const uuidRegex = /{{\s[a-f0-9-]+\s}}/g const found = newmarkdown.match(uuidRegex) @@ -810,7 +775,7 @@ const RunWorkflow = (defaultprops) => { } if (runWorkflow) { - loadInputWorkflowData(uuid, responseJson) + loadInputWorkflowData(uuid, workflow) } } @@ -819,6 +784,53 @@ const RunWorkflow = (defaultprops) => { } } + if (workflow.status !== "WAITING") { + setMessage("Already answered. You may close this window (3).") + } + } + + const getWorkflow = (workflow_id, selectedNode) => { + setRealtimeMarkdown("") + + const url = `${backendUrl}/api/v1/workflows/${workflow_id}` + fetch(url, { + method: "GET", + headers: { + "Content-Type": "application/json", + Accept: "application/json", + }, + credentials: "include", + }) + .then((response) => { + if (response.status !== 200) { + console.log("Status not 200 for workflows :O!"); + } + + if ((response.status === 401 || response.status === 403) && authorization === undefined || authorization === null || authorization.length === 0) { + toast(`This form is not available to you. If you think this is an error, please contact ${supportEmail} with the URL.`) + } + + return response.json() + }) + .then((responseJson) => { + // Not sure why this is necessary. + if (responseJson.isValid === undefined) { + responseJson.isValid = true; + } + + if (responseJson.errors === undefined) { + responseJson.errors = []; + } + + if (responseJson.actions === undefined || responseJson.actions === null) { + responseJson.actions = []; + } + + if (responseJson.triggers === undefined || responseJson.triggers === null) { + responseJson.triggers = []; + } + + setupSourcenode(responseJson, selectedNode) handleExecutionLoader() @@ -914,7 +926,7 @@ const RunWorkflow = (defaultprops) => { } // Just use this one? - var url = execution_id !== undefined && authorization !== undefined ? `${globalUrl}/api/v1/orgs/${orgId}?reference_execution=${execution_id}&authorization=${authorization}` : `${globalUrl}/api/v1/orgs/${orgId}`; + var url = execution_id !== undefined && authorization !== undefined ? `${backendUrl}/api/v1/orgs/${orgId}?reference_execution=${execution_id}&authorization=${authorization}` : `${backendUrl}/api/v1/orgs/${orgId}`; getWorkflows() loadForms(orgId) @@ -977,7 +989,7 @@ const RunWorkflow = (defaultprops) => { return } - fetch(globalUrl + "/api/v1/streams/results", { + fetch(backendUrl + "/api/v1/streams/results", { method: "POST", headers: { "Content-Type": "application/json", @@ -989,6 +1001,8 @@ const RunWorkflow = (defaultprops) => { .then((response) => { if (response.status !== 200) { console.log("Status not 200 for stream results :O!"); + + toast.warn("Error getting results.. Please try again or contact support@shuffler.io if this persists.") } return response.json(); @@ -998,6 +1012,13 @@ const RunWorkflow = (defaultprops) => { return } + if (execution_id !== undefined && execution_id !== null && authorization !== undefined && authorization !== null && execution_id.length > 0 && authorization.length > 0 && (workflow.id === undefined || workflow.id === null || workflow.id.length === 0) && responseJson.workflow !== undefined && responseJson.workflow !== null) { + + setupSourcenode(responseJson.workflow, sourceNode) + setWorkflow(responseJson.workflow) + } + + if (replaceMarkdown === true) { if (responseJson.result.length > 0) { // Set local storage for the workflow id @@ -1460,7 +1481,12 @@ const RunWorkflow = (defaultprops) => { (answer !== undefined && answer !== null) || message !== "" ? null : - Runtime Argument + {foundSourcenode !== undefined && foundSourcenode !== null ? + "Add Note" + : + "Runtime Argument" + } +
{ : null*/} : - ((answer !== undefined && answer !== null) || (foundSourcenode !== undefined && foundSourcenode !== null)) ? + (foundSourcenode !== undefined && foundSourcenode !== null) ? {disabledButtons && message.length > 0 ? - {message}. You may close this window. + {message} : - {disabledButtons ? "Answered. You may close this window." : ""} + {disabledButtons ? "Already answered. You may close this window." : ""} } diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 1a7f57b1..ed7e5add 100644 --- a/functions/onprem/orborus/go.mod +++ b/functions/onprem/orborus/go.mod @@ -10,7 +10,7 @@ require ( github.com/docker/docker v28.3.3+incompatible github.com/docker/go-connections v0.5.0 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.9.14 + github.com/shuffle/shuffle-shared v0.9.15 k8s.io/api v0.33.1 k8s.io/apimachinery v0.33.1 ) diff --git a/functions/onprem/orborus/go.sum b/functions/onprem/orborus/go.sum index 2496c612..3818c38f 100644 --- a/functions/onprem/orborus/go.sum +++ b/functions/onprem/orborus/go.sum @@ -328,8 +328,8 @@ github.com/sendgrid/sendgrid-go v3.16.1+incompatible h1:zWhTmB0Y8XCDzeWIm2/BIt1G github.com/sendgrid/sendgrid-go v3.16.1+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.9.14 h1:POkTHO+bByuv8HiKuCMSGgtpDKk86ISr6ooLG8vQfuE= -github.com/shuffle/shuffle-shared v0.9.14/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw= +github.com/shuffle/shuffle-shared v0.9.15 h1:Gc7c0pbWG6nHWSTkcfAnKgQgWCWfc6aDQ/BIu20z6bM= +github.com/shuffle/shuffle-shared v0.9.15/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw= 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= diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 719a7226..a6435b65 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -3937,7 +3937,6 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string, } identifier := "shuffle-workers" - if isKubernetes == "true" { if shuffle.IsRunningInCluster() { log.Printf("[INFO] Running in Kubernetes cluster") diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index ded2c196..0821bc1c 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -11,7 +11,7 @@ require ( github.com/docker/docker v28.3.3+incompatible github.com/gorilla/mux v1.8.1 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.9.14 + github.com/shuffle/shuffle-shared v0.9.15 github.com/shuffle/singul v0.0.16 k8s.io/api v0.33.1 k8s.io/apimachinery v0.33.1 diff --git a/functions/onprem/worker/go.sum b/functions/onprem/worker/go.sum index 2834c50b..0df9cc57 100644 --- a/functions/onprem/worker/go.sum +++ b/functions/onprem/worker/go.sum @@ -330,8 +330,8 @@ github.com/sendgrid/sendgrid-go v3.16.1+incompatible h1:zWhTmB0Y8XCDzeWIm2/BIt1G github.com/sendgrid/sendgrid-go v3.16.1+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.9.14 h1:POkTHO+bByuv8HiKuCMSGgtpDKk86ISr6ooLG8vQfuE= -github.com/shuffle/shuffle-shared v0.9.14/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw= +github.com/shuffle/shuffle-shared v0.9.15 h1:Gc7c0pbWG6nHWSTkcfAnKgQgWCWfc6aDQ/BIu20z6bM= +github.com/shuffle/shuffle-shared v0.9.15/go.mod h1:PhDEizuz4SmJaSmy0+yrFWwD1mXVUsy8/knKlrqF1qw= github.com/shuffle/singul v0.0.16 h1:dW+0Mln9R1aUJ0fjikpWcxbjoQWqJHxe4kSxh2tQN5E= github.com/shuffle/singul v0.0.16/go.mod h1:LYkp320A6gsoPlYbXUM+WvEPUVAuutlSsqnVKyRy4gs= github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 069910f5..c6b82823 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -304,7 +304,7 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason shutdownData, err := json.Marshal(workflowExecution) if err == nil { sendResult(workflowExecution, shutdownData) - log.Printf("[WARNING][%s] Sent shutdown update with %d results and result value %s", workflowExecution.ExecutionId, len(workflowExecution.Results), reason) + //log.Printf("[WARNING][%s] Sent shutdown update with %d results and result value %s", workflowExecution.ExecutionId, len(workflowExecution.Results), reason) } else { log.Printf("[WARNING][%s] Failed to send update: %s", workflowExecution.ExecutionId, err) } @@ -413,7 +413,6 @@ func deployk8sApp(image string, identifier string, env []string) error { } log.Printf("[DEBUG] Deploying k8s app with identifier %s to namespace %s", identifier, kubernetesNamespace) - deployport, err := strconv.Atoi(os.Getenv("SHUFFLE_APP_EXPOSED_PORT")) if err != nil { deployport = 80 @@ -2805,7 +2804,7 @@ func sendSelfRequest(actionResult shuffle.ActionResult) { if err != nil { log.Printf("[ERROR][%s] Failed reading body: %s", actionResult.ExecutionId, err) } else { - log.Printf("[DEBUG][%s] NEWRESP (from backend - 2): %s", actionResult.ExecutionId, string(body)) + log.Printf("[DEBUG][%s] Sent update to backend - 2: %s", actionResult.ExecutionId, string(body)) } } } @@ -2865,7 +2864,7 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { if err != nil { log.Printf("[ERROR][%s] Failed reading body: %s", workflowExecution.ExecutionId, err) } else { - log.Printf("[DEBUG][%s] NEWRESP (from backend): %s", workflowExecution.ExecutionId, string(body)) + log.Printf("[DEBUG][%s] Sent request to backend: %s", workflowExecution.ExecutionId, string(body)) } } } @@ -3081,12 +3080,9 @@ func getAvailablePort() (net.Listener, error) { listener, err := net.Listen("tcp", ":0") if err != nil { log.Printf("[WARNING] Failed to assign port by default. Defaulting to 5001") - //return ":5001" return nil, err } - //defer listener.Close() - return listener, nil //return fmt.Sprintf(":%d", port) } @@ -3181,7 +3177,7 @@ func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) { } /*** STARTREMOVE ***/ -func deploySwarmService(dockercli *dockerclient.Client, name, image string, deployport int, retry bool) error { +func deploySwarmService(dockercli *dockerclient.Client, name, image string, deployport int, inputReplicas int64, retry bool) error { log.Printf("[DEBUG] Deploying service for %s to swarm on port %d", name, deployport) //containerName := fmt.Sprintf("shuffle-worker-%s", parsedUuid) @@ -3242,6 +3238,15 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl log.Printf("[DEBUG] SHUFFLE_APP_REPLICAS set to value %#v. Trying to overwrite default (%d/node)", scaleReplicas, replicas) } + // Max scale as well + if inputReplicas > 0 && inputReplicas < 100 { + if replicas != uint64(inputReplicas) { + log.Printf("[DEBUG] Overwriting replicas to %d/node as inputReplicas is set to %d", inputReplicas, inputReplicas) + } + + replicas = uint64(inputReplicas) + } + cnt, err := findActiveSwarmNodes(dockercli) if err != nil { log.Printf("[ERROR] Unable to find active swarm nodes: %s", err) @@ -3382,10 +3387,20 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl // Retry deploying the service (once) if !retry { - return deploySwarmService(dockercli, name, image, deployport, true) + return deploySwarmService(dockercli, name, image, deployport, -1, true) } } + // For port mapping. + if strings.Contains(fmt.Sprintf("%s", err), "InvalidArgument") && strings.Contains(fmt.Sprintf("%s", err), "is already in use") { + //log.Printf("\n\n[WARNING] Port %d is already allocated. Trying to deploy on next port.\n\n", deployport) + + // Random sleep 1-4 seconds + time.Sleep(time.Duration(rand.Intn(4)+1) * time.Second) + + return deploySwarmService(dockercli, name, image, deployport+1, -1, retry) + } + log.Printf("[DEBUG] Failed deploying %s with image %s: %s", name, image, err) return err } @@ -3399,6 +3414,11 @@ func deploySwarmService(dockercli *dockerclient.Client, name, image string, depl func findAppInfo(image, name string, redeploy bool) (int, error) { + // Sleep between 0 and 1.5 second - ensures deployments have a higher + // chance of being successful + time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond) + + highest := baseport exposedPort := -1 @@ -3464,6 +3484,7 @@ func findAppInfo(image, name string, redeploy bool) (int, error) { continue } + // This seems to have concurrency issues portMappings[service.Spec.Annotations.Name] = int(endpoint.PublishedPort) if int(endpoint.PublishedPort) > highest { highest = int(endpoint.PublishedPort) @@ -3480,10 +3501,11 @@ func findAppInfo(image, name string, redeploy bool) (int, error) { } if redeploy { + log.Printf("Found it! Service: %s with image %s on port %d", name, image, exposedPort) // Remove the service and redeploy it. // There are cases where the service doesn't update properly - // Check when the last update happened. If it was within the last 5 minutes, skip - if int(time.Since(service.UpdatedAt).Seconds()) > 600 { + // Check when the last update happened. If it was within the last few minutes, skip + if int(time.Since(service.UpdatedAt).Seconds()) > 60 { log.Printf("[INFO] Attempting redeploy of app %s with image %s since it is more than 10 minutes since last attempt with failure.", name, image) @@ -3495,12 +3517,16 @@ func findAppInfo(image, name string, redeploy bool) (int, error) { log.Printf("[ERROR] Failed auto-removing service %s: %s", name, err) } else { log.Printf("[INFO] Auto-removed service %s successfully (rebuild due to redeploy).", name) - time.Sleep(10 * time.Second) + + // Sleep between 8 and 12 seconds + time.Sleep(time.Duration(rand.Intn(4)+8) * time.Second) + replicas := service.Spec.Mode.Replicated.Replicas err = deploySwarmService( dockercli, name, image, exposedPort, + int64(*replicas), false, ) if err != nil { @@ -3510,7 +3536,7 @@ func findAppInfo(image, name string, redeploy bool) (int, error) { } } } else { - //log.Printf("[INFO] NOT redeploying service %s since it was updated less than 10 minutes ago.", name) + //log.Printf("[INFO] NOT redeploying service %s since it was updated less than 3 minutes ago.", name) } } @@ -3538,7 +3564,7 @@ func findAppInfo(image, name string, redeploy bool) (int, error) { } highest += 1 - err = deploySwarmService(dockercli, name, image, highest, false) + err = deploySwarmService(dockercli, name, image, highest, -1, false) if err != nil { log.Printf("[WARNING] NOT Found service: %s. error: %s", name, err) return highest, err @@ -3900,7 +3926,7 @@ func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, // Has some issues with loading when running multiple workers and such. func baseDeploy() { - var cli *dockerclient.Client + //var cli *dockerclient.Client //var err error if isKubernetes != "true" { @@ -3956,14 +3982,19 @@ func baseDeploy() { env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))) } - identifier := fmt.Sprintf("%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId) - if strings.Contains(identifier, " ") { - identifier = strings.ReplaceAll(identifier, " ", "-") - } + identifier := fmt.Sprintf("%s_%s", appname, appversion) + //identifier := fmt.Sprintf("%s_%s_%s_%s", appname, appversion, action.ID, workflowExecution.ExecutionId) + //if strings.Contains(identifier, " ") { + // identifier = strings.ReplaceAll(identifier, " ", "-") + //} //deployApp(cli, value, identifier, env, workflowExecution, action) log.Printf("[DEBUG] Deploying app with identifier %s to ensure basic apps are available from the get-go", identifier) - go deployApp(cli, value, identifier, env, workflowExecution, action) + + //findAppInfo("frikky/shuffle:http_1.4.0", "http_1-4-0", true) + go findAppInfo(value, identifier, false) + + //go deployApp(cli, value, identifier, env, workflowExecution, action) //err := deployApp(cli, value, identifier, env, workflowExecution, action) //if err != nil { // log.Printf("[DEBUG] Failed deploying app %s: %s", value, err) @@ -4283,12 +4314,10 @@ func checkStandaloneRun() { // Initial loop etc func main() { - // Testing swarm auto-replacements. - //findAppInfo("frikky/shuffle:shuffle-ai_1.0.0", "shuffle-ai_1-0-0", true) - //findAppInfo("frikky/shuffle:shuffle-ai_1.0.0", "singul_1-0-0", true) + // Testing swarm auto-replacements. This also tests ports + // in rapid succession checkStandaloneRun() - if os.Getenv("DEBUG") == "true" { debug = true @@ -4299,9 +4328,10 @@ func main() { /*** STARTREMOVE ***/ if os.Getenv("SHUFFLE_SWARM_CONFIG") == "run" || os.Getenv("SHUFFLE_SWARM_CONFIG") == "swarm" { logsDisabled = "true" + os.Setenv("SHUFFLE_LOGS_DISABLED", "true") } - /*** ENDREMOVE ***/ + // Elasticsearch necessary to ensure we'ren ot running with Datastore configurations for minimal/maximal data sizes // Recursive import kind of :) _, err := shuffle.RunInit(*shuffle.GetDatastore(), *shuffle.GetStorage(), "", "worker", true, "elasticsearch", false, 0) @@ -4759,7 +4789,7 @@ func runWebserver(listener net.Listener) { //log.Fatal(http.Serve(listener, nil)) - log.Printf("[DEBUG] NEW webserver setup") + log.Printf("[DEBUG] NEW webserver setup. Port: %s", listener.Addr().String()) http.Handle("/", r) srv := http.Server{