diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod
index 6822373e..6aa6d2e0 100644
--- a/backend/go-app/go.mod
+++ b/backend/go-app/go.mod
@@ -4,7 +4,7 @@ go 1.23.0
toolchain go1.23.7
-replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
+//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared
require (
cloud.google.com/go/datastore v1.15.0
@@ -20,7 +20,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.32
+ github.com/shuffle/shuffle-shared v0.8.33
golang.org/x/crypto v0.36.0
google.golang.org/api v0.176.1
google.golang.org/grpc v1.68.1
diff --git a/backend/go-app/go.sum b/backend/go-app/go.sum
index 01efec32..a011691e 100644
--- a/backend/go-app/go.sum
+++ b/backend/go-app/go.sum
@@ -333,8 +333,8 @@ github.com/sendgrid/sendgrid-go v3.14.0+incompatible h1:KDSasSTktAqMJCYClHVE94Fc
github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
-github.com/shuffle/shuffle-shared v0.8.32 h1:RizHZ4lZg9wQj3RQvJMgh+1SAMszwhYfcco3aT716RQ=
-github.com/shuffle/shuffle-shared v0.8.32/go.mod h1:NruHSAscDsW595wpK2r7MeHPGspUEKRNvBpcN1iGbHI=
+github.com/shuffle/shuffle-shared v0.8.33 h1:XEiaUf2ykAPku9lX7JUXFDRBSropqiT0L2p4RdqVawE=
+github.com/shuffle/shuffle-shared v0.8.33/go.mod h1:NruHSAscDsW595wpK2r7MeHPGspUEKRNvBpcN1iGbHI=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
diff --git a/backend/go-app/main.go b/backend/go-app/main.go
index 6df286b9..dda3cfd2 100755
--- a/backend/go-app/main.go
+++ b/backend/go-app/main.go
@@ -3543,7 +3543,8 @@ func handleCloudJob(job shuffle.CloudSyncJob) error {
return err
}
- redirectDomain := "localhost:5001"
+ backendPort := os.Getenv("BACKEND_PORT")
+ redirectDomain := fmt.Sprintf("localhost:%s", backendPort)
redirectUrl := fmt.Sprintf("http://%s/api/v1/triggers/outlook/register", redirectDomain)
outlookClient, _, err := shuffle.GetOutlookClient(ctx, "", hook.OauthToken, redirectUrl)
if err != nil {
@@ -4199,12 +4200,13 @@ func runInitEs(ctx context.Context) {
return func() {
log.Printf("[INFO] Running schedule for cleaning up or re-running unfinished workflows in %d environments.", len(environments))
+ backendPort := os.Getenv("BACKEND_PORT")
for _, environment := range environments {
// Allowed without PROXY management as it's localhost
// client := shuffle.GetExternalClient(syncUrl)
httpClient := &http.Client{}
- url := fmt.Sprintf("http://localhost:5001/api/v1/environments/%s/stop", environment)
+ url := fmt.Sprintf("http://localhost:%s/api/v1/environments/%s/stop", backendPort, environment)
req, err := http.NewRequest(
"GET",
url,
@@ -4231,7 +4233,7 @@ func runInitEs(ctx context.Context) {
}
log.Printf("[DEBUG] Successfully ran workflow cleanup request for %s. Body: %s", environment, string(respBody))
- url = fmt.Sprintf("http://localhost:5001/api/v1/environments/%s/rerun", environment)
+ url = fmt.Sprintf("http://localhost:%s/api/v1/environments/%s/rerun", backendPort, environment)
req, err = http.NewRequest(
"GET",
url,
@@ -4256,7 +4258,7 @@ func runInitEs(ctx context.Context) {
log.Printf("[ERROR] Failed setting respbody %s", err)
continue
}
- log.Printf("[DEBUG] Successfully ran workflow RERUN request for %s. Body: %s", environment, string(respBody))
+ log.Printf("[DEBUG] Ran workflow RERUN request for %s with the response. Body: %s", environment, string(respBody))
}
}
}
@@ -5373,6 +5375,7 @@ func main() {
if innerPort == "" {
log.Printf("[DEBUG] Running on %s:5001", hostname)
log.Fatal(http.ListenAndServe(":5001", nil))
+ os.Setenv("BACKEND_PORT", "5001")
} else {
log.Printf("[DEBUG] Running on %s:%s", hostname, innerPort)
log.Fatal(http.ListenAndServe(fmt.Sprintf(":%s", innerPort), nil))
diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go
index 317595c3..b3aa3537 100755
--- a/backend/go-app/walkoff.go
+++ b/backend/go-app/walkoff.go
@@ -22,9 +22,6 @@ import (
dockerclient "github.com/docker/docker/client"
"github.com/docker/docker/api/types/image"
-
- //gyaml "github.com/ghodss/yaml"
-
"github.com/h2non/filetype"
uuid "github.com/satori/go.uuid"
@@ -36,14 +33,6 @@ import (
"github.com/go-git/go-git/v5/storage/memory"
"github.com/go-git/go-git/v5/plumbing"
http2 "github.com/go-git/go-git/v5/plumbing/transport/http"
- //http2 "gopkg.in/src-d/go-git.v5/plumbing/transport/http"
- //http2 "github.com/go-git/go-git/plumbing/transport/http"
-
- //"github.com/gorilla/websocket"
- //"google.golang.org/appengine"
- //"google.golang.org/appengine/memcache"
- //"cloud.google.com/go/firestore"
- // "google.golang.org/api/option"
gyaml "github.com/ghodss/yaml"
)
@@ -207,8 +196,6 @@ func handleGetWorkflowqueueConfirm(resp http.ResponseWriter, request *http.Reque
return
}
- log.Printf("Queue confirm: %#v. Queue: %#v", removeExecutionRequests.Data, id)
-
// remove items from DB
parsedId := strings.ReplaceAll(fmt.Sprintf("workflowqueue-%s", id), " ", "-")
ids := []string{}
diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx
index db55ee92..0ff57718 100755
--- a/frontend/src/views/AngularWorkflow.jsx
+++ b/frontend/src/views/AngularWorkflow.jsx
@@ -21288,20 +21288,18 @@ const AngularWorkflow = (defaultprops) => {
-
-
-
+
@@ -21641,7 +21639,6 @@ const AngularWorkflow = (defaultprops) => {
>
-
{
@@ -21651,9 +21648,8 @@ const AngularWorkflow = (defaultprops) => {
stop()
}}
>
- See more runs
+ Back to all runs
-
../../../../shuffle-shared
+//replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared
require (
github.com/docker/docker v27.5.0+incompatible
github.com/docker/go-connections v0.5.0
github.com/satori/go.uuid v1.2.0
- github.com/shuffle/shuffle-shared v0.8.19
+ github.com/shuffle/shuffle-shared v0.8.33
k8s.io/api v0.30.2
k8s.io/apimachinery v0.30.2
)
diff --git a/functions/onprem/orborus/go.sum b/functions/onprem/orborus/go.sum
index 39b13f42..6b5c6bd0 100644
--- a/functions/onprem/orborus/go.sum
+++ b/functions/onprem/orborus/go.sum
@@ -301,8 +301,8 @@ github.com/sendgrid/sendgrid-go v3.14.0+incompatible h1:KDSasSTktAqMJCYClHVE94Fc
github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
-github.com/shuffle/shuffle-shared v0.8.19 h1:HXqU62sPhVzv9MeJnA5ZpPYwwbitVz1EtMMABbY3t74=
-github.com/shuffle/shuffle-shared v0.8.19/go.mod h1:NruHSAscDsW595wpK2r7MeHPGspUEKRNvBpcN1iGbHI=
+github.com/shuffle/shuffle-shared v0.8.33 h1:XEiaUf2ykAPku9lX7JUXFDRBSropqiT0L2p4RdqVawE=
+github.com/shuffle/shuffle-shared v0.8.33/go.mod h1:NruHSAscDsW595wpK2r7MeHPGspUEKRNvBpcN1iGbHI=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go
index b5a3272b..b990c032 100755
--- a/functions/onprem/orborus/orborus.go
+++ b/functions/onprem/orborus/orborus.go
@@ -56,7 +56,7 @@ import (
"k8s.io/apimachinery/pkg/util/intstr"
)
-// Starts jobs in bulk, so this could be increased
+// Starts jobs in bulk, so this could be increased or decreased based on who the user is
var sleepTime = 2
// Making it work on low-end machines even during busy times :)
@@ -84,6 +84,7 @@ var appServiceAccountName = os.Getenv("SHUFFLE_APP_SERVICE_ACCOUNT_NAME")
// var baseimagename = "shuffle/shuffle"
var baseimagename = os.Getenv("SHUFFLE_BASE_IMAGE_NAME")
var baseimageregistry = os.Getenv("SHUFFLE_BASE_IMAGE_REGISTRY")
+
//var baseimagetagsuffix = os.Getenv("SHUFFLE_BASE_IMAGE_TAG_SUFFIX")
// Used for cloud with auth
@@ -431,19 +432,19 @@ func deployServiceWorkers(image string) {
}
/*
- isMemcachedRunning, err := checkMemcached(ctx, dockercli)
- if err != nil {
- log.Printf("[ERROR] Failed checking memcached: %s", err)
- }
- if isMemcachedRunning == false {
- log.Printf("[ERROR] Memcached is not running. Will try to deploy it.")
- deployMemcached(dockercli)
- }
+ isMemcachedRunning, err := checkMemcached(ctx, dockercli)
+ if err != nil {
+ log.Printf("[ERROR] Failed checking memcached: %s", err)
+ }
+ if isMemcachedRunning == false {
+ log.Printf("[ERROR] Memcached is not running. Will try to deploy it.")
+ deployMemcached(dockercli)
+ }
- ip := "shuffle-cache"
- if len(os.Getenv("SHUFFLE_MEMCACHED")) == 0 {
- os.Setenv("SHUFFLE_MEMCACHED", fmt.Sprintf("%s:11211", ip))
- }
+ ip := "shuffle-cache"
+ if len(os.Getenv("SHUFFLE_MEMCACHED")) == 0 {
+ os.Setenv("SHUFFLE_MEMCACHED", fmt.Sprintf("%s:11211", ip))
+ }
*/
defaultNetworkAttach := false
@@ -699,7 +700,6 @@ func deployServiceWorkers(image string) {
if err == nil {
log.Printf("[DEBUG] Successfully deployed workers with %d replica(s) on %d node(s)", replicas, cnt)
- //time.Sleep(time.Duration(10) * time.Second)
//log.Printf("[DEBUG] Servicecreate request: %#v %#v", service, err)
} else {
if !strings.Contains(fmt.Sprintf("%s", err), "Already Exists") && !strings.Contains(fmt.Sprintf("%s", err), "is already in use by service") {
@@ -743,9 +743,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
// Remove the image
handled := []string{}
- log.Printf("[DEBUG] Removing existing image (s): %s. Waiting 30 seconds before starting to ensure backend has the latest images built and ready to distribute.", images)
- //time.Sleep(time.Duration(30) * time.Second)
-
+ log.Printf("[DEBUG] Removing existing image (s): %s. Waiting 30 seconds before starting to ensure backend has the latest images built and ready to distribute.", images)
newImages := []string{}
for _, curimage := range strings.Split(images, ",") {
curimage = strings.TrimSpace(curimage)
@@ -763,7 +761,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
// Force remove the current image to avoid cached layers
if swarmConfig == "run" || swarmConfig == "swarm" {
_, err := dockercli.ImageRemove(ctx, curimage, image.RemoveOptions{
- Force: true,
+ Force: true,
PruneChildren: true,
})
@@ -831,7 +829,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
if !strings.Contains(fmt.Sprintf("%s", resp), "error") {
break
} else {
- found = true
+ found = true
log.Printf("[ERROR] Failed updating service %s with the new image %s: %s. Resp: %#v", service.Spec.Annotations.Name, image, err, resp)
}
}
@@ -1234,7 +1232,6 @@ func deployK8sWorker(image string, identifier string, env []string) error {
func deployWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error {
-
if len(os.Getenv("REGISTRY_URL")) > 0 && os.Getenv("REGISTRY_URL") != "" {
env = append(env, fmt.Sprintf("REGISTRY_URL=%s", os.Getenv("REGISTRY_URL")))
}
@@ -1294,7 +1291,6 @@ func deployWorker(image string, identifier string, env []string, executionReques
}
}
-
//var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG")
parsedUuid := uuid.NewV4()
@@ -1306,7 +1302,7 @@ func deployWorker(image string, identifier string, env []string, executionReques
if isKubernetes != "true" {
hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
- if strings.ToLower(cleanupEnv) != "false" {
+ if strings.ToLower(cleanupEnv) == "true" {
hostConfig.AutoRemove = true
}
}
@@ -1379,32 +1375,32 @@ func deployWorker(image string, identifier string, env []string, executionReques
log.Printf("[ERROR][%s] Failed to start worker container in environment '%s': %s", executionRequest.ExecutionId, environment, err)
return err
} else {
- log.Printf("[INFO][%s] Worker Container created (2). Environment %s: docker logs %s", executionRequest.ExecutionId, environment, cont.ID)
+ log.Printf("[INFO][%s] Worker Container created (2). Runtime Location '%s': docker logs -f %s", executionRequest.ExecutionId, environment, cont.ID)
}
stats, err := dockercli.ContainerInspect(ctx, cont.ID)
if err != nil {
log.Printf("[WARNING][%s] Failed checking worker '%s': %s", executionRequest.ExecutionId, cont.ID, err)
- return nil
+ return nil
}
containerStatus := stats.ContainerJSONBase.State.Status
if containerStatus != "running" {
log.Printf("[ERROR][%s] Status of %s is %s. Should be running. Contact support@shuffler.io if this persists.", executionRequest.ExecutionId, cont.ID, containerStatus)
}
- /*
- err = stopWorker(containerName)
- if err != nil {
- log.Printf("Failed stopping worker %s", execution.ExecutionId)
- return nil
- }
+ /*
+ err = stopWorker(containerName)
+ if err != nil {
+ log.Printf("Failed stopping worker %s", execution.ExecutionId)
+ return nil
+ }
- err = deployWorker(dockercli, workerImage, containerName, env)
- if err != nil {
- log.Printf("Failed executing worker %s in state %s", execution.ExecutionId, containerStatus)
- return nil
+ err = deployWorker(dockercli, workerImage, containerName, env)
+ if err != nil {
+ log.Printf("Failed executing worker %s in state %s", execution.ExecutionId, containerStatus)
+ return nil
+ }
}
- }
*/
} else {
log.Printf("[INFO][%s] New Worker created. Environment %s: docker logs %s", executionRequest.ExecutionId, environment, cont.ID)
@@ -1452,8 +1448,8 @@ func initializeImages() {
}
if baseimageregistry == "" {
+ //baseimageregistry = "ghcr.io" // Github
baseimageregistry = "docker.io" // Dockerhub
- baseimageregistry = "ghcr.io" // Github
if len(os.Getenv("REGISTRY_URL")) > 0 {
baseimageregistry = os.Getenv("REGISTRY_URL")
@@ -1462,14 +1458,14 @@ func initializeImages() {
}
os.Setenv("SHUFFLE_BASE_IMAGE_REGISTRY", baseimageregistry)
-
+
log.Printf("[WARNING] Setting baseimageregistry to %#v", baseimageregistry)
}
if baseimagename == "" {
// FIXME: This is probably the problem for image names tbh
+ //baseimagename = "shuffle" // Github (ghcr.io)
baseimagename = "frikky/shuffle" // Dockerhub
- baseimagename = "shuffle" // Github (ghcr.io)
os.Setenv("SHUFFLE_BASE_IMAGE_NAME", baseimagename)
log.Printf("[WARNING] Setting baseimagename to %#v", baseimagename)
@@ -1477,19 +1473,18 @@ func initializeImages() {
log.Printf("[DEBUG] Setting swarm config to %#v. Default is empty.", swarmConfig)
- // FIXME: Shuffle Worker vs Apps != same
- newWorker := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
+ // This is now always static
+ newWorker := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)
if len(newWorkerImage) > 0 {
newWorker = newWorkerImage
}
- // check whether they are the same first
-
- if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") != "true" {
+ // Check whether they are the same first
+ if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" {
+ log.Printf("[DEBUG] Skipping image download as SHUFFLE_AUTO_IMAGE_DOWNLOAD is set to false")
+ } else {
images := []string{
fmt.Sprintf("frikky/shuffle:app_sdk"),
- fmt.Sprintf("shuffle/shuffle:app_sdk"),
- fmt.Sprintf("%s/%s/shuffle-app_sdk:%s", baseimageregistry, baseimagename, appSdkVersion),
newWorker,
}
@@ -1510,8 +1505,6 @@ func initializeImages() {
log.Printf("[DEBUG] Successfully downloaded and built %s", image)
}
}
- } else {
- log.Printf("[DEBUG] Skipping image download as SHUFFLE_AUTO_IMAGE_DOWNLOAD is set to true")
}
}
@@ -2005,7 +1998,7 @@ func main() {
log.Printf("[INFO] Setting up Docker environment. Downloading worker and App SDK!")
initializeImages()
- workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
+ workerImage := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)
if len(newWorkerImage) > 0 {
workerImage = newWorkerImage
}
@@ -2116,6 +2109,7 @@ func main() {
// Marshal and set body
orborusStats := getOrborusStats(ctx)
+
pipelinePayload, pipelineerr := sendPipelineHealthStatus()
if pipelineerr != nil {
@@ -2515,9 +2509,23 @@ func main() {
executionIds = append(executionIds, execution.ExecutionId)
} else {
log.Printf("[WARNING][%s] Failed to deploy: %s", execution.ExecutionId, err)
+
if strings.Contains(err.Error(), "already exists") {
toBeRemoved.Data = append(toBeRemoved.Data, execution)
executionIds = append(executionIds, execution.ExecutionId)
+ } else if strings.Contains(err.Error(), "No such image") {
+ // Download the image
+
+ if isKubernetes == "true" {
+ log.Printf("[DEBUG] Skipping image pull of '%s' because Kubernetes does it in realtime instead", workerImage)
+ } else {
+ log.Printf("[DEBUG] Re-pulling image %s as it doesn't exist, and is necessary for worker to run (autofix)", workerImage)
+ pullOptions := image.PullOptions{}
+ _, err = dockercli.ImagePull(ctx, workerImage, pullOptions)
+ if err != nil {
+ log.Printf("[ERROR] Failed to pull image %s: %s", workerImage, err)
+ }
+ }
}
}
}
@@ -3208,10 +3216,28 @@ func deletePipeline(pipelineId string) error {
func listPipelines() ([]shuffle.PipelineInfo, error) {
responseData := shuffle.PipelineInfoWrapper{}
+ if tenzirDisabled {
+ return responseData.Pipelines, errors.New("Tenzir is disabled")
+ }
+
var reqBody []byte
url := fmt.Sprintf("%s/api/v0/pipeline/list", pipelineUrl)
- resp, err := http.Post(url, "application/json", bytes.NewBuffer(reqBody))
+ client := http.Client{
+ Timeout: 2 * time.Second,
+ }
+ req, err := http.NewRequest(
+ "POST",
+ url,
+ bytes.NewBuffer(reqBody),
+ )
+
+ if err != nil {
+ return responseData.Pipelines, err
+ }
+
+ req.Header.Set("Content-Type", "application/json")
+ resp, err := client.Do(req)
if err != nil {
return responseData.Pipelines, err
}
@@ -3438,6 +3464,10 @@ func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
Pipelines: []shuffle.PipelineInfoMini{},
}
+ if tenzirDisabled {
+ return pipelinePayload, nil
+ }
+
// To not spam down the list API too much
randint := rand.Intn(5)
if len(pipelines) == 0 || randint == 0 {
@@ -3460,10 +3490,6 @@ func sendPipelineHealthStatus() (shuffle.LakeConfig, error) {
pipelinePayload.Pipelines = pipelines
}
- if tenzirDisabled {
- return pipelinePayload, nil
- }
-
err := deployTenzirNode()
if err != nil {
if (!strings.Contains(err.Error(), "SHUFFLE_SKIP_PIPELINES") && !strings.Contains(err.Error(), "Kubernetes not implemented for Tenzir node")) && !strings.Contains(err.Error(), "Tenzir Node is already running") && !strings.Contains(err.Error(), "docker daemon") {
@@ -3800,8 +3826,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
if err != nil {
log.Printf("[ERROR] Failed creating worker request: %s", err)
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
- workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
-
+ workerImage := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)
if len(newWorkerImage) > 0 {
workerImage = newWorkerImage
}
@@ -3825,8 +3850,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
log.Printf("[ERROR] Error running worker request to %s (1): %s", streamUrl, err)
if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") {
- workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion)
-
+ workerImage := fmt.Sprintf("ghcr.io/shuffle/shuffle-worker:%s", workerVersion)
if len(newWorkerImage) > 0 {
workerImage = newWorkerImage
}
diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod
index 23f27d64..eacebda6 100644
--- a/functions/onprem/worker/go.mod
+++ b/functions/onprem/worker/go.mod
@@ -8,7 +8,7 @@ require (
github.com/docker/docker v27.5.0+incompatible
github.com/gorilla/mux v1.8.1
github.com/satori/go.uuid v1.2.0
- github.com/shuffle/shuffle-shared v0.8.31
+ github.com/shuffle/shuffle-shared v0.8.33
k8s.io/api v0.30.2
k8s.io/apimachinery v0.30.2
k8s.io/client-go v0.30.2
diff --git a/functions/onprem/worker/go.sum b/functions/onprem/worker/go.sum
index 88c9ba70..360e7230 100644
--- a/functions/onprem/worker/go.sum
+++ b/functions/onprem/worker/go.sum
@@ -294,8 +294,8 @@ github.com/sendgrid/sendgrid-go v3.14.0+incompatible h1:KDSasSTktAqMJCYClHVE94Fc
github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 h1:n661drycOFuPLCN3Uc8sB6B/s6Z4t2xvBgU1htSHuq8=
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
-github.com/shuffle/shuffle-shared v0.8.31 h1:APO/BkBxiP9Hn/Fa7SsESRD6ws9E5dBfIzS57pd3MvA=
-github.com/shuffle/shuffle-shared v0.8.31/go.mod h1:NruHSAscDsW595wpK2r7MeHPGspUEKRNvBpcN1iGbHI=
+github.com/shuffle/shuffle-shared v0.8.33 h1:XEiaUf2ykAPku9lX7JUXFDRBSropqiT0L2p4RdqVawE=
+github.com/shuffle/shuffle-shared v0.8.33/go.mod h1:NruHSAscDsW595wpK2r7MeHPGspUEKRNvBpcN1iGbHI=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go
index 5dc288f8..2a3a4301 100644
--- a/functions/onprem/worker/worker.go
+++ b/functions/onprem/worker/worker.go
@@ -105,9 +105,9 @@ var window = shuffle.NewTimeWindow(10 * time.Second)
// Images to be autodeployed in the latest version of Shuffle.
var autoDeploy = map[string]string{
- "http:1.4.0": "frikky/shuffle:http_1.4.0",
- "shuffle-tools:1.2.0": "frikky/shuffle:shuffle-tools_1.2.0",
- "shuffle-subflow:1.1.0": "frikky/shuffle:shuffle-subflow_1.1.0",
+ "http:1.4.0": "frikky/shuffle:http_1.4.0",
+ "shuffle-tools:1.2.0": "frikky/shuffle:shuffle-tools_1.2.0",
+ "shuffle-subflow:1.1.0": "frikky/shuffle:shuffle-subflow_1.1.0",
// "shuffle-tools-fork:1.0.0": "frikky/shuffle:shuffle-tools-fork_1.0.0",
}
@@ -299,7 +299,7 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
}
// Might not be necessary because of cleanupEnv hostconfig autoremoval
- if cleanupEnv == "true" && (os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") {
+ if strings.ToLower(cleanupEnv) == "true" && (os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") {
/*
ctx := context.Background()
dockercli, err := dockerclient.NewEnvClient()
@@ -884,7 +884,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
}
// Removing because log extraction should happen first
- if cleanupEnv == "true" {
+ if strings.ToLower(cleanupEnv) == "true" {
hostConfig.AutoRemove = true
}
@@ -1247,6 +1247,7 @@ func getWorkerURLs() ([]string, error) {
}
func askOtherWorkersToDownloadImage(image string) {
+ // Why wouldn't it happen on swarm? Hmm
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
return
}
@@ -1489,10 +1490,6 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")))
}
- // Fixes issue:
- // standard_go init_linux.go:185: exec user process caused "argument list too long"
- // https://devblogs.microsoft.com/oldnewthing/20100203-00/?p=15083
-
// FIXME: Ensure to NEVER do this anymore
// This potentially breaks too much stuff. Better to have the app poll the data.
_ = executionData
@@ -1516,9 +1513,11 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion),
}
- // If cleanup is set, it should run for efficiency
+ // This is the weirdest shit ever looking back at
+ // Needs optimization lol
+
pullOptions := image.PullOptions{}
- if cleanupEnv == "true" {
+ if strings.ToLower(cleanupEnv) == "true" {
err = deployApp(dockercli, images[0], identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
if strings.Contains(err.Error(), "exited prematurely") {