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

Merge pull request #1804 from Shuffle/nightly

Browse Source 
Issues for app scaling and building fixed
This commit is contained in:
Frikky
2025-09-02 13:44:11 +02:00
committed by GitHub
parent 9ed6351a9a 9f3070b65c
commit 1b63e6cba6
12 changed files with 317 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/go-app/docker.go
+12 -10
View File
@@ -57,14 +57,11 @@ func getParsedTar(tw *tar.Writer, baseDir, extra string) error {
switch mode := fi.Mode(); {
case mode.IsDir():
// do directory recursion
//log.Printf("DIR: %s", file)
// Append "src" as extra here
filenamesplit := strings.Split(file, "/")
// Cross-platform path handling for tar entries
filenamesplit := strings.Split(filepath.ToSlash(file), "/")
filename := fmt.Sprintf("%s%s/", extra, filenamesplit[len(filenamesplit)-1])
tmpExtra := fmt.Sprintf(filename)
//log.Printf("TmpExtra: %s", tmpExtra)
err = getParsedTar(tw, file, tmpExtra)
if err != nil {
log.Printf("Directory parse issue: %s", err)
@@ -86,9 +83,8 @@ func getParsedTar(tw *tar.Writer, baseDir, extra string) error {
return err
}
filenamesplit := strings.Split(file, "/")
filenamesplit := strings.Split(filepath.ToSlash(file), "/")
filename := fmt.Sprintf("%s%s", extra, filenamesplit[len(filenamesplit)-1])
//log.Printf("Filename: %s", filename)
tarHeader := &tar.Header{
Name: filename,
Size: int64(len(readFile)),
@@ -486,16 +482,22 @@ func buildImage(tags []string, dockerfileLocation string) error {
}
log.Printf("[INFO] Docker Tags: %s", tags)
dockerfileSplit := strings.Split(dockerfileLocation, "/")
log.Printf("[DEBUG] Dockerfile location: %s", dockerfileLocation)
// Convert to forward slashes for consistent handling across OS
normalizedPath := filepath.ToSlash(dockerfileLocation)
dockerfileSplit := strings.Split(normalizedPath, "/")
// Create a buffer
buf := new(bytes.Buffer)
tw := tar.NewWriter(buf)
defer tw.Close()
// Use the directory part of the dockerfile path
baseDir := strings.Join(dockerfileSplit[0:len(dockerfileSplit)-1], "/")
// Builds the entire folder into buf
err = getParsedTar(tw, baseDir, "")
// Builds the entire folder into buf using OS-specific path
err = getParsedTar(tw, filepath.FromSlash(baseDir), "")
if err != nil {
log.Printf("[ERROR] Tar issue during app build: %s", err)
}
backend/go-app/go.mod
+1 -1
View File
@@ -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.16
github.com/shuffle/singul v0.0.16
golang.org/x/crypto v0.40.0
google.golang.org/api v0.236.0
backend/go-app/go.sum
+2 -2
View File
@@ -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.16 h1:B3Dp3h2f62JdWmqQ1kTu7DwJY6HCOMrxxSojAeuhI9k=
github.com/shuffle/shuffle-shared v0.9.16/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=
backend/go-app/main.go
+29
View File
@@ -3326,6 +3326,35 @@ func buildSwaggerApp(resp http.ResponseWriter, body []byte, user shuffle.User, s
log.Printf("[INFO] Successfully stitched ZIPFILE for %s", identifier)
// Copy baseline Dockerfile to build directory
dockerfileSource := "../app_gen/python-lib/baseline/Dockerfile"
dockerfileDestination := fmt.Sprintf("%s/Dockerfile", basePath)
// Read and copy the baseline Dockerfile
dockerfileContent, err := ioutil.ReadFile(dockerfileSource)
if err != nil {
log.Printf("[ERROR] Failed to read baseline Dockerfile: %s", err)
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false, "reason": "Failed to read baseline Dockerfile"}`))
return
}
err = ioutil.WriteFile(dockerfileDestination, dockerfileContent, 0644)
if err != nil {
log.Printf("[ERROR] Failed to copy Dockerfile to build directory: %s", err)
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false, "reason": "Failed to copy Dockerfile"}`))
return
}
// Verify the Dockerfile was created
if _, err := os.Stat(dockerfileDestination); os.IsNotExist(err) {
log.Printf("[ERROR] Dockerfile does not exist at destination: %s", dockerfileDestination)
resp.WriteHeader(500)
resp.Write([]byte(`{"success": false, "reason": "Dockerfile was not created properly"}`))
return
}
// 4. Build the image locally.
// FIXME: Should be moved to a local docker registry
dockerLocation := fmt.Sprintf("%s/Dockerfile", basePath)
frontend/src/components/EnvironmentTab.jsx
+71 -41
View File
@@ -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) => {
<Tooltip title={
<Typography variant="body1" style={{margin: 10, }}>
{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."
:
<span>IP / label: {environment?.running_ip?.split(":")[0]}. May stay running up to a minute after stopping Orborus.</span>
:
@@ -1102,7 +1100,7 @@ const EnvironmentTab = memo((props) => {
<br />
<br />
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"}
</Typography>
} placement="top">
<Typography
@@ -1117,7 +1115,36 @@ const EnvironmentTab = memo((props) => {
}}
variant="body2"
>
{environment.Type !== "cloud" &&
{environment.Type === "cloud" ?
<Chip
key={index}
style={{
color: green,
borderColor: green,
}}
label={"Running"}
onClick={() => {
//handleChipClick
}}
variant="outlined"
color="primary"
/>
:
selectedOrganization.id !== undefined && environment?.org_id !== selectedOrganization.id ?
<Chip
key={index}
style={{
color: yellow,
borderColor: yellow,
}}
label={"N/A"}
onClick={() => {
//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) => {
<ListItemText
primary={
selectedOrganization.id !== undefined && environment?.org_id !== selectedOrganization.id ?
"N/A"
:
environment.licensed ? (
<Tooltip title="Scale configured (auto on cloud)" placement="top">
<CheckCircleIcon style={{ color: "#4caf50" }} />
</Tooltip>
) : (
<Tooltip
title="In Verbose mode. Set SHUFFLE_SWARM_CONFIG=run to Scale. This will not be as verbose. Details: https://shuffler.io/docs/configuration#scaling-shuffle"
placement="top"
>
<a
href="/docs/configuration#scaling-shuffle"
target="_blank"
rel="noopener noreferrer"
>
<CancelIcon style={{ color: "#f85a3e" }} />
</a>
</Tooltip>
)
}
style={{
minWidth: 60,
marginLeft: 20,
overflow: "hidden",
whiteSpace: "normal",
wordWrap: "break-word",
padding: 8,
display: "table-cell",
}}
/>
<CheckCircleIcon style={{ color: "#4caf50" }} />
</Tooltip>
) : (
<Tooltip
title="In Verbose mode. Set SHUFFLE_SWARM_CONFIG=run to Scale. This will not be as verbose. Details: https://shuffler.io/docs/configuration#scaling-shuffle"
placement="top"
>
<a
href="/docs/configuration#scaling-shuffle"
target="_blank"
rel="noopener noreferrer"
>
<CancelIcon style={{ color: "#f85a3e" }} />
</a>
</Tooltip>
)
}
style={{
minWidth: 60,
marginLeft: 20,
overflow: "hidden",
whiteSpace: "normal",
wordWrap: "break-word",
padding: 8,
display: "table-cell",
}}
/>
<ListItemText
primary={
frontend/src/views/RunWorkflow.jsx
+115 -89
View File
@@ -85,6 +85,12 @@ const RunWorkflow = (defaultprops) => {
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 :
<span>
Runtime Argument
{foundSourcenode !== undefined && foundSourcenode !== null ?
"Add Note"
:
"Runtime Argument"
}
<div style={{marginBottom: 5}}>
<TextField
color="primary"
@@ -1506,17 +1532,17 @@ const RunWorkflow = (defaultprops) => {
: null*/}
</span>
:
((answer !== undefined && answer !== null) || (foundSourcenode !== undefined && foundSourcenode !== null)) ?
(foundSourcenode !== undefined && foundSourcenode !== null) ?
<span style={{marginTop: 20, }}>
{disabledButtons && message.length > 0 ?
<Typography variant="body1" style={{textAlign: "center", marginTop: 30, marginBottom: 20, }}>
{message}. You may close this window.
{message}
</Typography>
:
<Fade in={true} timeout={2500}>
<Typography variant="body1" style={{textAlign: "center", marginTop: 30, marginBottom: 20, }}>
{disabledButtons ? "Answered. You may close this window." : ""}
{disabledButtons ? "Already answered. You may close this window." : ""}
</Typography>
</Fade>
}
functions/onprem/orborus/go.mod
+1 -1
View File
@@ -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
)
functions/onprem/orborus/go.sum
+2 -2
View File
@@ -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=
functions/onprem/orborus/orborus.go
+22 -6
View File
@@ -60,7 +60,7 @@ var sleepTime = 2
// Making it work on low-end machines even during busy times :)
// May cause some things to run slowly
var maxConcurrency = 7
var maxConcurrency = 25
// Timeout if something rashes
var workerTimeoutEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_TIMEOUT")
@@ -81,6 +81,7 @@ var workerContainerSecurityContext = os.Getenv("SHUFFLE_WORKER_CONTAINER_SECURIT
var appServiceAccountName = os.Getenv("SHUFFLE_APP_SERVICE_ACCOUNT_NAME")
var appPodSecurityContext = os.Getenv("SHUFFLE_APP_POD_SECURITY_CONTEXT")
var appContainerSecurityContext = os.Getenv("SHUFFLE_APP_CONTAINER_SECURITY_CONTEXT")
var debug = os.Getenv("DEBUG") == "true"
// var baseimagename = "docker.pkg.github.com/shuffle/shuffle"
// var baseimagename = "ghcr.io/frikky"
@@ -218,6 +219,10 @@ func skipCheckInCleanup(name string) bool {
}
func cleanupExistingNodes(ctx context.Context) error {
if cleanupEnv == "false" {
log.Printf("[INFO] Skipping cleanup of existing workers as CLEANUP is set to false. This should be auto-discovered during executions then instead.")
return nil
}
if isKubernetes == "true" {
// Cleanup all workers created by orborus and all apps created by workers.
@@ -2173,6 +2178,12 @@ func main() {
client := shuffle.GetExternalClient(baseUrl)
fullUrl := fmt.Sprintf("%s/api/v1/workflows/queue", baseUrl)
// Increases default concurrency to 50 for swarm
if maxConcurrency < 50 && (swarmConfig == "run" || swarmConfig == "swarm") {
fullUrl += "?amount=50"
}
if isKubernetes == "true" {
log.Printf("[INFO] Finished configuring kubernetes environment. Connecting to %s", fullUrl)
} else {
@@ -3926,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")
@@ -3938,7 +3948,9 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
// Specific to debugging
if len(workerServerUrl) == 0 {
log.Printf("[INFO] Using default worker server url as previous is invalid: %s", streamUrl)
if debug {
log.Printf("[INFO] Using default worker server url as previous is invalid: %s. Swapping to shuffle-workers:33333", streamUrl)
}
}
streamUrl = fmt.Sprintf("http://shuffle-workers:33333/api/v1/execute")
@@ -3993,7 +4005,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
newresp, err := client.Do(req)
if err != nil {
// Connection refused?
log.Printf("[ERROR] Error running worker request to %s (1): %s", streamUrl, err)
log.Printf("[ERROR][%s] Error running worker request to %s (1): %s", workflowExecution.ExecutionId, streamUrl, err)
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
workerImage := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)
@@ -4026,7 +4038,11 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
log.Printf("[WARNING] POTENTIAL error running worker request (2) - status code is %d for %s, not 200. Body: %s", newresp.StatusCode, streamUrl, string(body))
// In case of old executions
if strings.Contains(string(body), "Bad status ") {
if strings.Contains(strings.ToLower(string(body)), "bad status ") {
return nil
}
if strings.Contains(strings.ToLower(string(body)), "no apps to handle") {
return nil
}
@@ -4040,7 +4056,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
debugCommand = fmt.Sprintf("kubectl logs -n %s deployment/shuffle-workers | grep %s", kubernetesNamespace, workflowExecution.ExecutionId)
}
log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s. DEBUGGING:\n%s", workflowExecution.ExecutionId, streamUrl, debugCommand)
log.Printf("[DEBUG][%s] Ran worker from requests. Worker URL: %s. DEBUGGING:\n%s", workflowExecution.ExecutionId, streamUrl, debugCommand)
return nil
}
functions/onprem/worker/go.mod
+1 -1
View File
@@ -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
functions/onprem/worker/go.sum
+2 -2
View File
@@ -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=
functions/onprem/worker/worker.go
+59 -27
View File
@@ -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
@@ -2022,7 +2021,9 @@ func executionInit(workflowExecution shuffle.WorkflowExecution) error {
}
if len(onpremApps) == 0 {
return errors.New(fmt.Sprintf("No apps to handle onprem (%s)", environment))
//return errors.New(fmt.Sprintf("No apps to handle onprem (%s)", environment))
log.Printf("[INFO][%s] No apps to handle onprem (%s). Returning 200 OK anyway", workflowExecution.ExecutionId, environment)
return nil
}
pullOptions := dockerimage.PullOptions{}
@@ -2803,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))
}
}
}
@@ -2863,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))
}
}
}
@@ -3079,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)
}
@@ -3179,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)
@@ -3240,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)
@@ -3380,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
}
@@ -3397,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
@@ -3462,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)
@@ -3478,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)
@@ -3493,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 {
@@ -3508,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)
}
}
@@ -3536,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
@@ -3898,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" {
@@ -3954,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)
@@ -4281,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
@@ -4297,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)
@@ -4757,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{