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

Worker fix

Browse Source
This commit is contained in:
Frikky
2024-08-20 23:37:21 +02:00
parent fe2e3105eb
commit af14b76ac5
4 changed files with 689 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/go-app/go.sum
+2
View File
@@ -465,6 +465,8 @@ github.com/shuffle/shuffle-shared v0.6.61 h1:+9CCLeZLiAVDgNRTkZxnIgz+FZ7UrEHez2B
github.com/shuffle/shuffle-shared v0.6.61/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
github.com/shuffle/shuffle-shared v0.6.62 h1:NWVjVbnNpm6osDHQS2NGqwAgWk1Fk3s5RKUO3P5n4js=
github.com/shuffle/shuffle-shared v0.6.62/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
github.com/shuffle/shuffle-shared v0.6.63 h1:eNQMpVhe/mAMxl61W9Wj6/Z4PrtPeEnbjvMtDdT1mqw=
github.com/shuffle/shuffle-shared v0.6.63/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
functions/onprem/worker/go.mod
+7 -5
View File
@@ -7,9 +7,9 @@ require (
github.com/gorilla/mux v1.8.1
github.com/satori/go.uuid v1.2.0
github.com/shuffle/shuffle-shared v0.6.63
k8s.io/api v0.30.0
k8s.io/apimachinery v0.30.0
k8s.io/client-go v0.30.0
k8s.io/api v0.30.2
k8s.io/apimachinery v0.30.2
k8s.io/client-go v0.30.2
)
require (
@@ -27,7 +27,7 @@ require (
github.com/algolia/algoliasearch-client-go/v3 v3.18.1 // indirect
github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874 // indirect
github.com/bradfitz/slice v0.0.0-20180809154707-2b758aa73013 // indirect
github.com/cloudflare/circl v1.3.3 // indirect
github.com/cloudflare/circl v1.3.7 // indirect
github.com/containerd/log v0.1.0 // indirect
github.com/cyphar/filepath-securejoin v0.2.4 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
@@ -38,7 +38,7 @@ require (
github.com/emirpasic/gods v1.18.1 // indirect
github.com/felixge/httpsnoop v1.0.4 // indirect
github.com/frikky/kin-openapi v0.41.0 // indirect
github.com/frikky/schemaless v0.0.11 // indirect
github.com/frikky/schemaless v0.0.13 // indirect
github.com/ghodss/yaml v1.0.0 // indirect
github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect
github.com/go-git/go-billy/v5 v5.5.0 // indirect
@@ -79,6 +79,8 @@ require (
github.com/pjbgf/sha1cd v0.3.0 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/sashabaranov/go-openai v1.19.2 // indirect
github.com/sendgrid/rest v2.6.9+incompatible // indirect
github.com/sendgrid/sendgrid-go v3.14.0+incompatible // indirect
github.com/sergi/go-diff v1.1.0 // indirect
github.com/skeema/knownhosts v1.2.1 // indirect
github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e // indirect
functions/onprem/worker/go.sum
+16
View File
@@ -97,6 +97,8 @@ github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMn
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
github.com/cloudflare/circl v1.3.3 h1:fE/Qz0QdIGqeWfnwq0RE0R7MI51s0M2E4Ga9kq5AEMs=
github.com/cloudflare/circl v1.3.3/go.mod h1:5XYMA4rFBvNIrhs50XuiBJ15vF2pZn4nnUKZrLbUZFA=
github.com/cloudflare/circl v1.3.7 h1:qlCDlTPz2n9fu58M0Nh1J/JzcFpfgkFHHX3O35r5vcU=
github.com/cloudflare/circl v1.3.7/go.mod h1:sRTcRWXGLrKw6yIGJ+l7amYJFfAXbZG0kBSc8r4zxgA=
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa h1:jQCWAUqqlij9Pgj2i/PB79y4KOPYVyFYdROxgaCwdTQ=
@@ -139,6 +141,8 @@ github.com/frikky/schemaless v0.0.9 h1:RzNLPkJq5c4nlm5iLiTndFcbeQxdMGJIj266wSGt2
github.com/frikky/schemaless v0.0.9/go.mod h1:mooDxY+D6weHjhKvjy3+IE9S7P4g4cpNnidkdRv/cHQ=
github.com/frikky/schemaless v0.0.11 h1:c4r6CJX30XI+SoJdT9RlUd9qYSQlx6hvwGRtsypu+uM=
github.com/frikky/schemaless v0.0.11/go.mod h1:mooDxY+D6weHjhKvjy3+IE9S7P4g4cpNnidkdRv/cHQ=
github.com/frikky/schemaless v0.0.13 h1:ARiN9V7wr2VZXAr9JK5wvTbyPgpGrgeiL1VhR5MlgaQ=
github.com/frikky/schemaless v0.0.13/go.mod h1:mooDxY+D6weHjhKvjy3+IE9S7P4g4cpNnidkdRv/cHQ=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
@@ -387,12 +391,18 @@ github.com/sashabaranov/go-openai v1.19.2 h1:+dkuCADSnwXV02YVJkdphY8XD9AyHLUWwk6
github.com/sashabaranov/go-openai v1.19.2/go.mod h1:lj5b/K+zjTSFxVLijLSTDZuP7adOgerWeFyZLUhAKRg=
github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
github.com/sendgrid/rest v2.6.9+incompatible h1:1EyIcsNdn9KIisLW50MKwmSRSK+ekueiEMJ7NEoxJo0=
github.com/sendgrid/rest v2.6.9+incompatible/go.mod h1:kXX7q3jZtJXK5c5qK83bSGMdV6tsOE70KbHoqJls4lE=
github.com/sendgrid/sendgrid-go v3.14.0+incompatible h1:KDSasSTktAqMJCYClHVE94Fcif2i7P7wzISv1sU6DUA=
github.com/sendgrid/sendgrid-go v3.14.0+incompatible/go.mod h1:QRQt+LX/NmgVEvmdRw0VT/QgUn499+iza2FnDca9fg8=
github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
github.com/shuffle/shuffle-shared v0.6.16 h1:dQBDRmb2Wgl3pEuewqjDvN6v6nUKr+1EvGSEja9zG6s=
github.com/shuffle/shuffle-shared v0.6.16/go.mod h1:HhQTn7xZZ69ZTc4EptO9OeNmgbKDyGlWAhFkUFUAHSA=
github.com/shuffle/shuffle-shared v0.6.27 h1:q4qZD6bGZFIvZ5Y10unGr3N3rZ7OryWyvvaGgANZJZU=
github.com/shuffle/shuffle-shared v0.6.27/go.mod h1:rWkh1eWdIx7OqQzJ1+JzF3Hck1X/Ty1WkUtjLrp+CU4=
github.com/shuffle/shuffle-shared v0.6.63 h1:eNQMpVhe/mAMxl61W9Wj6/Z4PrtPeEnbjvMtDdT1mqw=
github.com/shuffle/shuffle-shared v0.6.63/go.mod h1:RAJiSFjmuKmijKTbbEf9A6Ojb+3/te7g71lED7JjPus=
github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
@@ -906,10 +916,16 @@ honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9
honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
k8s.io/api v0.30.0 h1:siWhRq7cNjy2iHssOB9SCGNCl2spiF1dO3dABqZ8niA=
k8s.io/api v0.30.0/go.mod h1:OPlaYhoHs8EQ1ql0R/TsUgaRPhpKNxIMrKQfWUp8QSE=
k8s.io/api v0.30.2 h1:+ZhRj+28QT4UOH+BKznu4CBgPWgkXO7XAvMcMl0qKvI=
k8s.io/api v0.30.2/go.mod h1:ULg5g9JvOev2dG0u2hig4Z7tQ2hHIuS+m8MNZ+X6EmI=
k8s.io/apimachinery v0.30.0 h1:qxVPsyDM5XS96NIh9Oj6LavoVFYff/Pon9cZeDIkHHA=
k8s.io/apimachinery v0.30.0/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc=
k8s.io/apimachinery v0.30.2 h1:fEMcnBj6qkzzPGSVsAZtQThU62SmQ4ZymlXRC5yFSCg=
k8s.io/apimachinery v0.30.2/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc=
k8s.io/client-go v0.30.0 h1:sB1AGGlhY/o7KCyCEQ0bPWzYDL0pwOZO4vAtTSh/gJQ=
k8s.io/client-go v0.30.0/go.mod h1:g7li5O5256qe6TYdAMyX/otJqMhIiGgTapdLchhmOaY=
k8s.io/client-go v0.30.2 h1:sBIVJdojUNPDU/jObC+18tXWcTJVcwyqS9diGdWHk50=
k8s.io/client-go v0.30.2/go.mod h1:JglKSWULm9xlJLx4KCkfLLQ7XwtlbflV6uFFSHTMgVs=
k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw=
k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 h1:BZqlfIlq5YbRMFko6/PM7FjZpUb45WallggurYhKGag=
functions/onprem/worker/worker.go
+404 -118
View File
@@ -3,7 +3,6 @@ package main
import (
"github.com/shuffle/shuffle-shared"
"bytes"
"context"
"encoding/json"
@@ -22,18 +21,21 @@ import (
"time"
"github.com/docker/docker/api/types"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/container"
"github.com/docker/docker/api/types/filters"
"github.com/docker/docker/api/types/mount"
dockerclient "github.com/docker/docker/client"
// This is for automatic removal of certain code :)
"github.com/gorilla/mux"
"github.com/satori/go.uuid"
uuid "github.com/satori/go.uuid"
//k8s deps
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
appsv1 "k8s.io/api/apps/v1"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/kubernetes"
)
@@ -77,6 +79,7 @@ var startAction string
//var allLogs map[string]string
//var containerIds []string
var downloadedImages []string
type ImageDownloadBody struct {
Image string `json:"image"`
}
@@ -87,7 +90,7 @@ type ImageRequest struct {
var finishedExecutions []string
var imagesDistributed []string
var imagedownloadTimeout = time.Second * 300
// Images to be autodeployed in the latest version of Shuffle.
var autoDeploy = map[string]string{
@@ -188,14 +191,12 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo
}
}
for _, trigger := range workflowExecution.Workflow.Triggers {
//log.Printf("[DEBUG] Found trigger %s", trigger.AppName)
if trigger.AppName != "User Input" && trigger.AppName != "Shuffle Workflow" && trigger.AppName != "shuffle-subflow" {
continue
}
// check if it has wait for results in params
wait := false
for _, param := range trigger.Parameters {
@@ -249,7 +250,6 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo
os.Exit(0)
}
break
}
@@ -271,7 +271,6 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo
return nil
}
// removes every container except itself (worker)
func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) {
log.Printf("[DEBUG][%s] Shutdown (%s) started with reason %#v. Result amount: %d. ResultsSent: %d, Send result: %#v, Parent: %#v", workflowExecution.ExecutionId, workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend, workflowExecution.ExecutionParent)
@@ -369,7 +368,7 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
//Finished shutdown (after %d seconds). ", sleepDuration)
// Allows everything to finish in subprocesses (apps)
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" && isKubernetes != "true" {
time.Sleep(time.Duration(sleepDuration) * time.Second)
os.Exit(3)
} else {
@@ -377,9 +376,10 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
}
}
// Deploys the internal worker whenever something happens
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, action shuffle.Action) error {
if isKubernetes == "true" {
func int32Ptr(i int32) *int32 { return &i }
// ** STARTREMOVE ***/
func deployk8sApp(image string, identifier string, env []string) error {
if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 {
kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
} else {
@@ -394,16 +394,24 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
}
}
// add to env
// fmt.Sprintf("SHUFFLE_APP_EXPOSED_PORT=%d", deployport),
// fmt.Sprintf("SHUFFLE_SWARM_CONFIG=%s", os.Getenv("SHUFFLE_SWARM_CONFIG")),
envMap["SHUFFLE_APP_EXPOSED_PORT"] = "80"
envMap["SHUFFLE_SWARM_CONFIG"] = os.Getenv("SHUFFLE_SWARM_CONFIG")
envMap["BASE_URL"] = "http://shuffle-workers:33333"
clientset, _, err := shuffle.GetKubernetesClient()
if err != nil {
log.Printf("[ERROR] Failed getting kubernetes: %s", err)
return err
}
str := strings.ToLower(identifier)
strSplit := strings.Split(str, "_")
value := strSplit[0]
value = strings.ReplaceAll(value, "_", "-")
// str := strings.ToLower(identifier)
// strSplit := strings.Split(str, "_")
// value := strSplit[0]
// value = strings.ReplaceAll(value, "_", "-")
value := identifier
// Checking if app is generated or not
localRegistry := os.Getenv("REGISTRY_URL")
@@ -443,44 +451,259 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
log.Printf("[DEBUG] Got kubernetes with namespace %#v to run image '%s'", kubernetesNamespace, image)
//fix naming convention
podUuid := uuid.NewV4().String()
podName := fmt.Sprintf("%s-%s", value, podUuid)
// podUuid := uuid.NewV4().String()
// podName := fmt.Sprintf("%s-%s", value, podUuid)
// replace identifier "_" with "-"
podName := strings.ReplaceAll(identifier, "_", "-")
pod := &corev1.Pod{
// pod := &corev1.Pod{
// ObjectMeta: metav1.ObjectMeta{
// Name: podName,
// Labels: map[string]string{
// "app": podName,
// // "executionId": workflowExecution.ExecutionId,
// },
// },
// Spec: corev1.PodSpec{
// RestartPolicy: "Never", // As a crash is not useful in this context
// // DNSPolicy: "Default",
// DNSPolicy: corev1.DNSClusterFirst,
// // NodeName: "worker1"
// Containers: []corev1.Container{
// {
// Name: value,
// Image: image,
// Env: buildEnvVars(envMap),
// // Pull if not available
// ImagePullPolicy: corev1.PullIfNotPresent,
// },
// },
// },
// }
// createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
// if err != nil {
// log.Printf("[ERROR] Failed creating pod: %v", err)
// // os.Exit(1)
// } else {
// log.Printf("[DEBUG] Created pod %#v in namespace %#v", createdPod.Name, kubernetesNamespace)
// }
// service := &corev1.Service{
// ObjectMeta: metav1.ObjectMeta{
// Name: identifier,
// },
// Spec: corev1.ServiceSpec{
// Selector: map[string]string{
// "app": podName,
// },
// Ports: []corev1.ServicePort{
// {
// Protocol: "TCP",
// Port: 80,
// TargetPort: intstr.FromInt(80),
// },
// },
// Type: corev1.ServiceTypeNodePort,
// },
// }
// _, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.TODO(), service, metav1.CreateOptions{})
// if err != nil {
// log.Printf("[ERROR] Failed creating service: %v", err)
// return err
// }
// use deployment instead of pod
// then expose a service similarly.
// number of replicas can be set to os.Getenv("SHUFFLE_SCALE_REPLICAS")
replicaNumberStr := os.Getenv("SHUFFLE_SCALE_REPLICAS")
replicaNumber := 1
if len(replicaNumberStr) > 0 {
tmpInt, err := strconv.Atoi(replicaNumberStr)
if err != nil {
log.Printf("[ERROR] %s is not a valid number for replication", replicaNumberStr)
} else {
replicaNumber = tmpInt
}
}
replicaNumberInt32 := int32(replicaNumber)
deployment := &appsv1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Name: podName,
},
Spec: appsv1.DeploymentSpec{
Replicas: int32Ptr(replicaNumberInt32),
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"app": podName,
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"app": "shuffle-app",
"executionId": workflowExecution.ExecutionId,
"app": podName,
},
},
Spec: corev1.PodSpec{
RestartPolicy: "Never", // As a crash is not useful in this context
DNSPolicy: "Default",
// NodeName: "worker1"
Containers: []corev1.Container{
{
Name: value,
Image: image,
Env: buildEnvVars(envMap),
// Pull if not available
ImagePullPolicy: corev1.PullIfNotPresent,
},
},
},
},
},
}
createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
_, err = clientset.AppsV1().Deployments(kubernetesNamespace).Create(context.Background(), deployment, metav1.CreateOptions{})
if err != nil {
log.Printf("[ERROR] Failed creating pod: %v", err)
// os.Exit(1)
} else {
log.Printf("[DEBUG] Created pod %#v in namespace %#v", createdPod.Name, kubernetesNamespace)
log.Printf("[ERROR] Failed creating deployment: %v", err)
return err
}
// kubectl expose deployment {podName} --type=NodePort --port=80 --target-port=80
service := &corev1.Service{
ObjectMeta: metav1.ObjectMeta{
Name: podName,
},
Spec: corev1.ServiceSpec{
Selector: map[string]string{
"app": podName,
},
Ports: []corev1.ServicePort{
{
Protocol: "TCP",
Port: 80,
TargetPort: intstr.FromInt(80),
},
},
Type: corev1.ServiceTypeNodePort,
},
}
_, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.TODO(), service, metav1.CreateOptions{})
if err != nil {
log.Printf("[ERROR] Failed creating service: %v", err)
return err
}
return nil
}
}
//** ENDREMOVE ***/
// Deploys the internal worker whenever something happens
func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, action shuffle.Action) error {
// if isKubernetes == "true" {
// if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 {
// kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
// } else {
// kubernetesNamespace = "default"
// }
// envMap := make(map[string]string)
// for _, envStr := range env {
// parts := strings.SplitN(envStr, "=", 2)
// if len(parts) == 2 {
// envMap[parts[0]] = parts[1]
// }
// }
// clientset, _, err := shuffle.GetKubernetesClient()
// if err != nil {
// log.Printf("[ERROR] Failed getting kubernetes: %s", err)
// return err
// }
// str := strings.ToLower(identifier)
// strSplit := strings.Split(str, "_")
// value := strSplit[0]
// value = strings.ReplaceAll(value, "_", "-")
// // Checking if app is generated or not
// localRegistry := os.Getenv("REGISTRY_URL")
// /*
// appDetails := strings.Split(image, ":")[1]
// appDetailsSplit := strings.Split(appDetails, "_")
// appName := strings.Join(appDetailsSplit[:len(appDetailsSplit)-1], "_")
// appVersion := appDetailsSplit[len(appDetailsSplit)-1]
// for _, app := range workflowExecution.Workflow.Actions {
// // log.Printf("[DEBUG] App: %s, Version: %s", appName, appVersion)
// // log.Printf("[DEBUG] Checking app %s with version %s", app.AppName, app.AppVersion)
// if app.AppName == appName && app.AppVersion == appVersion {
// if app.Generated == true {
// log.Printf("[DEBUG] Generated app, setting local registry")
// image = fmt.Sprintf("%s/%s", localRegistry, image)
// break
// } else {
// log.Printf("[DEBUG] Not generated app, setting shuffle registry")
// }
// }
// }
// */
// if len(localRegistry) == 0 && len(os.Getenv("SHUFFLE_BASE_IMAGE_REGISTRY")) > 0 {
// localRegistry = os.Getenv("SHUFFLE_BASE_IMAGE_REGISTRY")
// }
// if len(localRegistry) > 0 && strings.Count(image, "/") <= 2 {
// log.Printf("[DEBUG] Using REGISTRY_URL %s", localRegistry)
// image = fmt.Sprintf("%s/%s", localRegistry, image)
// } else {
// if strings.Count(image, "/") <= 2 {
// image = fmt.Sprintf("frikky/shuffle:%s", image)
// }
// }
// log.Printf("[DEBUG] Got kubernetes with namespace %#v to run image '%s'", kubernetesNamespace, image)
// //fix naming convention
// podUuid := uuid.NewV4().String()
// podName := fmt.Sprintf("%s-%s", value, podUuid)
// pod := &corev1.Pod{
// ObjectMeta: metav1.ObjectMeta{
// Name: podName,
// Labels: map[string]string{
// "app": "shuffle-app",
// "executionId": workflowExecution.ExecutionId,
// },
// },
// Spec: corev1.PodSpec{
// RestartPolicy: "Never", // As a crash is not useful in this context
// // DNSPolicy: "Default",
// DNSPolicy: corev1.DNSClusterFirst,
// // NodeName: "worker1"
// Containers: []corev1.Container{
// {
// Name: value,
// Image: image,
// Env: buildEnvVars(envMap),
// // Pull if not available
// ImagePullPolicy: corev1.PullIfNotPresent,
// },
// },
// },
// }
// createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
// if err != nil {
// log.Printf("[ERROR] Failed creating pod: %v", err)
// // os.Exit(1)
// } else {
// log.Printf("[DEBUG] Created pod %#v in namespace %#v", createdPod.Name, kubernetesNamespace)
// }
// return nil
// }
// form basic hostConfig
ctx := context.Background()
@@ -542,7 +765,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
// Get environment for certificates
volumeBinds := []string{}
volumeBindString:= os.Getenv("SHUFFLE_VOLUME_BINDS")
volumeBindString := os.Getenv("SHUFFLE_VOLUME_BINDS")
if len(volumeBindString) > 0 {
volumeBindSplit := strings.Split(volumeBindString, ",")
for _, volumeBind := range volumeBindSplit {
@@ -583,7 +806,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
Env: env,
}
// Checking as late as possible, just in case.
newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
_, err := shuffle.GetCache(ctx, newExecId)
@@ -618,8 +840,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
}
func cleanupKubernetesExecution(clientset *kubernetes.Clientset, workflowExecution shuffle.WorkflowExecution, namespace string) error {
workerName := fmt.Sprintf("worker-%s", workflowExecution.ExecutionId)
// workerName := fmt.Sprintf("worker-%s", workflowExecution.ExecutionId)
labelSelector := fmt.Sprintf("app=shuffle-app,executionId=%s", workflowExecution.ExecutionId)
podList, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
@@ -637,11 +858,11 @@ func cleanupKubernetesExecution(clientset *kubernetes.Clientset, workflowExecuti
log.Printf("App %s in namespace %s deleted.", pod.Name, namespace)
}
podErr := clientset.CoreV1().Pods(namespace).Delete(context.TODO(), workerName, metav1.DeleteOptions{})
if podErr != nil {
return fmt.Errorf("[ERROR] failed to delete the worker %s in namespace %s: %v", workerName, namespace, podErr)
}
log.Printf("[DEBUG] %s in namespace %s deleted.", workerName, namespace)
// podErr := clientset.CoreV1().Pods(namespace).Delete(context.TODO(), workerName, metav1.DeleteOptions{})
// if podErr != nil {
// return fmt.Errorf("[ERROR] failed to delete the worker %s in namespace %s: %v", workerName, namespace, podErr)
// }
// log.Printf("[DEBUG] %s in namespace %s deleted.", workerName, namespace)
return nil
}
@@ -824,6 +1045,48 @@ func removeIndex(s []string, i int) []string {
func getWorkerURLs() ([]string, error) {
workerUrls := []string{}
if isKubernetes == "true" {
workerUrls = append(workerUrls, "http://shuffle-workers:33333")
// workerUrls = append(workerUrls, "http://192.168.29.16:33333")
// get service "shuffle-workers" "Endpoints"
// serviceName := "shuffle-workers"
// clientset, _, err := shuffle.GetKubernetesClient()
// if err != nil {
// log.Println("[ERROR] Failed to get Kubernetes client:", err)
// return workerUrls, err
// }
// services, err := clientset.CoreV1().Services("default").List(context.Background(), metav1.ListOptions{})
// if err != nil {
// log.Println("[ERROR] Failed to list services:", err)
// return workerUrls, err
// }
// for _, service := range services.Items {
// if service.Name == serviceName {
// endpoints, err := clientset.CoreV1().Endpoints("default").Get(context.Background(), serviceName, metav1.GetOptions{})
// if err != nil {
// log.Println("[ERROR] Failed to get endpoints for service:", err)
// return workerUrls, err
// }
// for _, subset := range endpoints.Subsets {
// for _, address := range subset.Addresses {
// for _, port := range subset.Ports {
// url := fmt.Sprintf("http://%s:%d", address.IP, port.Port)
// workerUrls = append(workerUrls, url)
// }
// }
// }
// }
// }
log.Printf("[DEBUG] Worker URLs for k8s: %#v", workerUrls)
return workerUrls, nil
}
// Create a new Docker client
cli, err := dockerclient.NewEnvClient()
if err != nil {
@@ -876,11 +1139,10 @@ func askOtherWorkersToDownloadImage(image string) {
return
}
if len(urls) < 2{
if len(urls) < 2 {
return
}
httpClient := &http.Client{}
distributed := false
for _, url := range urls {
@@ -936,16 +1198,22 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
return
}
startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId)
dockercli, err := dockerclient.NewEnvClient()
var dockercli *dockerclient.Client
var err error
if isKubernetes != "true" {
dockercli, err = dockerclient.NewEnvClient()
if err != nil {
log.Printf("[ERROR] Unable to create docker client (3): %s", err)
return
}
}
if isKubernetes == "true" {
defer dockercli.Close()
}
for _, action := range relevantActions {
appname := action.AppName
@@ -976,7 +1244,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
//executed = append(executed, action.ID)
// FIXME - check whether it's running locally yet too
// take care of auto clean up later on for k8s
if isKubernetes != "true" {
stats, err := dockercli.ContainerInspect(context.Background(), identifier)
if err != nil || stats.ContainerJSONBase.State.Status != "running" {
// REMOVE
@@ -984,7 +1253,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
log.Printf("[DEBUG][%s] Docker Container Status: %s, should kill: %s", workflowExecution.ExecutionId, stats.ContainerJSONBase.State.Status, identifier)
err = removeContainer(identifier)
if err != nil {
log.Printf("Error killing container: %s", err)
log.Printf("[ERROR] Error killing container: %s", err)
}
} else {
//log.Printf("WHAT TO DO HERE?: %s", err)
@@ -993,6 +1262,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
//log.Printf("
continue
}
}
if len(action.Parameters) == 0 {
action.Parameters = []shuffle.WorkflowAppActionParameter{}
@@ -1093,7 +1363,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
@@ -1121,8 +1390,6 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion),
}
// If cleanup is set, it should run for efficiency
pullOptions := types.ImagePullOptions{}
if cleanupEnv == "true" {
@@ -1134,7 +1401,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
return
}
err := downloadDockerImageBackend(&http.Client{Timeout: 60 * time.Second}, image)
err := downloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, image)
executed := false
if err == nil {
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
@@ -1247,13 +1514,15 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
}
log.Printf("[DEBUG][%s] Failed deploy. Downloading image %s: %s", workflowExecution.ExecutionId, image, err)
err := downloadDockerImageBackend(&http.Client{Timeout: 60 * time.Second}, image)
err := downloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, image)
executed := false
if err == nil {
log.Printf("[DEBUG] Downloaded image %s from backend (CLEANUP)", image)
//err = deployApp(dockercli, image, identifier, env, workflow, action)
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Err: %s", err)
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (40)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
@@ -1268,6 +1537,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
image = images[2]
err = deployApp(dockercli, image, identifier, env, workflowExecution, action)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Err: %s", err)
if strings.Contains(err.Error(), "exited prematurely") {
log.Printf("[DEBUG] Shutting down (11)")
shutdown(workflowExecution, action.ID, fmt.Sprintf("%s", err.Error()), true)
@@ -1276,6 +1546,11 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
log.Printf("[WARNING] Failed deploying image THREE TIMES. Attempting to download %s as last resort from backend and dockerhub: %s", image, err)
if isKubernetes == "true" {
log.Printf("[ERROR] Image %s doesn't exist. Returning error for now")
return
}
reader, err := dockercli.ImagePull(context.Background(), image, pullOptions)
if err != nil && !strings.Contains(err.Error(), "Conflict. The container name") {
log.Printf("[ERROR] Failed getting %s. The couldn't be find locally, AND is missing.", image)
@@ -1625,13 +1900,11 @@ func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.Workflow
}
}
if workflowExecution.Status == "WAITING" && workflowExecution.ExecutionSource != "default" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" {
log.Printf("[INFO][%s] Workflow execution is waiting. Exiting worker, as backend will restart it.", workflowExecution.ExecutionId)
shutdown(workflowExecution, "", "", true)
}
log.Printf("[INFO][%s] (2) Status: %s, Results: %d, actions: %d. Userinput: %#v", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra, hasUserinput)
return errors.New("Subflow status not found yet")
}
@@ -1898,7 +2171,6 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
if request.Body == nil {
log.Printf("[WARNING] (2) No body in request for workflowqueue")
resp.WriteHeader(http.StatusBadRequest)
return
}
@@ -1974,9 +2246,9 @@ func handleWorkflowQueue(resp http.ResponseWriter, request *http.Request) {
log.Printf("[DEBUG][%s] Action: Received, Label: '%s', Action: '%s', Status: %s, Run status: %s, Extra=Retry:%d", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.AppName, actionResult.Status, workflowExecution.Status, retries)
//results = append(results, actionResult)
//log.Printf("[INFO][%s] Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", workflowExecution.ExecutionId, action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters))
//log.Printf("[DEBUG][%s] In workflowQueue with transaction", workflowExecution.ExecutionId)
// results = append(results, actionResult)
// log.Printf("[INFO][%s] Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", workflowExecution.ExecutionId, action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters))
// log.Printf("[DEBUG][%s] In workflowQueue with transaction", workflowExecution.ExecutionId)
runWorkflowExecutionTransaction(ctx, 0, workflowExecution.ExecutionId, actionResult, resp)
}
@@ -2003,7 +2275,6 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl
resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Execution is not executing, but %s"}`, workflowExecution.Status)))
}
log.Printf("[DEBUG][%s] Shutting down (35)", workflowExecution.ExecutionId)
// Force sending result
@@ -2228,9 +2499,9 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) {
log.Printf("[DEBUG][%s] NEWRESP (from backend): %s", workflowExecution.ExecutionId, string(body))
}
}
}
}
func validateFinished(workflowExecution shuffle.WorkflowExecution) bool {
func validateFinished(workflowExecution shuffle.WorkflowExecution) bool {
ctx := context.Background()
newexec, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId)
@@ -2376,7 +2647,11 @@ func getAvailablePort() (net.Listener, error) {
func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener {
hostname = getLocalIP()
if isKubernetes == "true" {
os.Setenv("WORKER_HOSTNAME", "shuffle-workers")
} else {
os.Setenv("WORKER_HOSTNAME", hostname)
}
// FIXME: This MAY not work because of speed between first
// container being launched and port being assigned to webserver
@@ -2506,9 +2781,9 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error {
log.Printf("[INFO] Successfully loaded image %s: %s", imageName, string(body))
return nil
}
}
func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) {
func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) {
ctx := context.Background()
nodes, err := dockercli.NodeList(ctx, types.NodeListOptions{})
if err != nil {
@@ -2550,7 +2825,7 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error {
// Runs data discovery
func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution) error {
parsedRequest := shuffle.OrborusExecutionRequest{
parsedRequest := shuffle.OrborusExecutionRequest{
Cleanup: cleanupEnv,
ExecutionId: workflowExecution.ExecutionId,
Authorization: workflowExecution.Authorization,
@@ -2563,21 +2838,21 @@ parsedRequest := shuffle.OrborusExecutionRequest{
BaseUrl: baseUrl,
Action: *action,
FullExecution: *workflowExecution,
}
// Sometimes makes it have the wrong data due to timing
}
// Sometimes makes it have the wrong data due to timing
// Specific for subflow to ensure worker matches the backend correctly
// Specific for subflow to ensure worker matches the backend correctly
parsedBaseurl := incomingUrl
if strings.Count(baseUrl, ":") >= 2 {
parsedBaseurl := incomingUrl
if strings.Count(baseUrl, ":") >= 2 {
baseUrlSplit := strings.Split(baseUrl, ":")
if len(baseUrlSplit) >= 3 {
parsedBaseurl = strings.Join(baseUrlSplit[0:2], ":")
//parsedRequest.BaseUrl = fmt.Sprintf("%s:33333", parsedBaseurl)
}
}
}
if len(parsedRequest.Url) == 0 {
if len(parsedRequest.Url) == 0 {
// Fixed callback url to the worker itself
if strings.Count(parsedBaseurl, ":") >= 2 {
parsedRequest.Url = parsedBaseurl
@@ -2589,17 +2864,17 @@ if len(parsedRequest.Url) == 0 {
}
//log.Printf("[DEBUG][%s] Should add a baseurl for the app to get back to: %s", workflowExecution.ExecutionId, parsedRequest.Url)
}
}
// Swapping because this was confusing during dev
// No real reason, just variable names
tmp := parsedRequest.Url
parsedRequest.Url = parsedRequest.BaseUrl
parsedRequest.BaseUrl = tmp
// Swapping because this was confusing during dev
// No real reason, just variable names
tmp := parsedRequest.Url
parsedRequest.Url = parsedRequest.BaseUrl
parsedRequest.BaseUrl = tmp
// Run with proper hostname, but set to shuffle-worker to avoid specific host target.
// This means running with VIP instead.
if len(hostname) > 0 {
// Run with proper hostname, but set to shuffle-worker to avoid specific host target.
// This means running with VIP instead.
if len(hostname) > 0 {
parsedRequest.BaseUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
//parsedRequest.BaseUrl = fmt.Sprintf("http://shuffle-workers:%d", baseport)
//log.Printf("[DEBUG][%s] Changing hostname to local hostname in Docker network for WORKER URL: %s", workflowExecution.ExecutionId, parsedRequest.BaseUrl)
@@ -2608,49 +2883,53 @@ if len(hostname) > 0 {
parsedRequest.BaseUrl = fmt.Sprintf("http://%s:%d", hostname, baseport)
//parsedRequest.Url = parsedRequest.BaseUrl
}
}
}
// Making sure to get the LATEST execution data
// This is due to cache timing issues
exec, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId)
if err == nil && len(exec.ExecutionId) > 0 {
// Making sure to get the LATEST execution data
// This is due to cache timing issues
exec, err := shuffle.GetWorkflowExecution(ctx, workflowExecution.ExecutionId)
if err == nil && len(exec.ExecutionId) > 0 {
parsedRequest.FullExecution = *exec
}
}
data, err := json.Marshal(parsedRequest)
if err != nil {
data, err := json.Marshal(parsedRequest)
if err != nil {
log.Printf("[ERROR] Failed marshalling worker request: %s", err)
return err
}
}
streamUrl := fmt.Sprintf("http://%s:%d/api/v1/run", appName, port)
//log.Printf("[DEBUG][%s] Worker URL: %s, Backend URL: %s, Target App: %s", workflowExecution.ExecutionId, parsedRequest.BaseUrl, parsedRequest.Url, streamUrl)
req, err := http.NewRequest(
if isKubernetes == "true" {
appName = strings.Replace(appName, "_", "-", -1)
}
streamUrl := fmt.Sprintf("http://%s:%d/api/v1/run", appName, port)
// log.Printf("[DEBUG][%s] Worker URL: %s, Backend URL: %s, Target App: %s", workflowExecution.ExecutionId, parsedRequest.BaseUrl, parsedRequest.Url, streamUrl)
req, err := http.NewRequest(
"POST",
streamUrl,
bytes.NewBuffer([]byte(data)),
)
)
// Checking as LATE as possible, ensuring we don't rerun what's already ran
//ctx = context.Background()
newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
_, err = shuffle.GetCache(ctx, newExecId)
if err == nil {
// Checking as LATE as possible, ensuring we don't rerun what's already ran
// ctx = context.Background()
newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
_, err = shuffle.GetCache(ctx, newExecId)
if err == nil {
log.Printf("[DEBUG] Result for %s already found (PRE REQUEST) - returning", newExecId)
return nil
}
}
cacheData := []byte("1")
err = shuffle.SetCache(ctx, newExecId, cacheData, 30)
if err != nil {
cacheData := []byte("1")
err = shuffle.SetCache(ctx, newExecId, cacheData, 30)
if err != nil {
log.Printf("[WARNING] Failed setting cache for action %s: %s", newExecId, err)
} else {
} else {
//log.Printf("[DEBUG][%s] Adding %s to cache (%#v)", workflowExecution.ExecutionId, newExecId, action.Name)
}
}
client := shuffle.GetExternalClient(streamUrl)
customTimeout := os.Getenv("SHUFFLE_APP_REQUEST_TIMEOUT")
if len(customTimeout) > 0 {
client := shuffle.GetExternalClient(streamUrl)
customTimeout := os.Getenv("SHUFFLE_APP_REQUEST_TIMEOUT")
if len(customTimeout) > 0 {
// convert to int
timeoutInt, err := strconv.Atoi(customTimeout)
if err != nil {
@@ -2659,10 +2938,10 @@ if len(customTimeout) > 0 {
log.Printf("[DEBUG] Setting client timeout to %d seconds for app request", timeoutInt)
client.Timeout = time.Duration(timeoutInt) * time.Second
}
}
}
newresp, err := client.Do(req)
if err != nil {
newresp, err := client.Do(req)
if err != nil {
// Another timeout issue here somewhere
// context deadline
if strings.Contains(fmt.Sprintf("%s", err), "context deadline exceeded") || strings.Contains(fmt.Sprintf("%s", err), "Client.Timeout exceeded") {
@@ -2724,13 +3003,17 @@ if err != nil {
// Has some issues with loading when running multiple workers and such.
func baseDeploy() {
var cli *dockerclient.Client
var err error
if isKubernetes != "true" {
cli, err := dockerclient.NewEnvClient()
if err != nil {
log.Printf("[ERROR] Unable to create docker client (3): %s", err)
return
}
defer cli.Close()
}
for key, value := range autoDeploy {
newNameSplit := strings.Split(key, ":")
@@ -2810,7 +3093,7 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe
if newresp.StatusCode != 200 {
log.Printf("[ERROR] %sStatusCode (1): %d", string(body), newresp.StatusCode)
time.Sleep(time.Duration(sleepTime) * time.Second)
return environments, errors.New(fmt.Sprintf("Bad status code: %d", newresp.StatusCode) )
return environments, errors.New(fmt.Sprintf("Bad status code: %d", newresp.StatusCode))
}
err = json.Unmarshal(body, &workflowExecution)
@@ -2850,7 +3133,7 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe
}
// Checks if a subflow is child of the startnode, as sub-subflows aren't working properly yet
childNodes := shuffle.FindChildNodes(workflowExecution, workflowExecution.Start, []string{}, []string{})
childNodes := shuffle.FindChildNodes(workflowExecution.Workflow, workflowExecution.Start, []string{}, []string{})
log.Printf("[DEBUG] Looking for subflow in %#v to check execution pattern as child of %s", childNodes, workflowExecution.Start)
subflowFound := false
for _, childNode := range childNodes {
@@ -2893,7 +3176,6 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe
// Set environment variable
//log.Printf("Before wait")
//wg := sync.WaitGroup{}
//wg.Add(1)
@@ -2944,7 +3226,11 @@ func main() {
log.Printf("[ERROR] Failed to run worker init: %s", err)
}
} else {
if isKubernetes != "true" {
log.Printf("[DEBUG] Ran init for worker to set up cache system. Docker version: %s", dockerApiVersion)
} else {
log.Printf("[DEBUG] Ran init for worker to set up cache system on Kubernetes")
}
}
log.Printf("[INFO] Setting up worker environment")
@@ -3270,7 +3556,6 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) {
return
}
for _, img := range images {
for _, tag := range img.RepoTags {
splitTag := strings.Split(tag, ":")
@@ -3283,7 +3568,7 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) {
possibleNames = append(possibleNames, fmt.Sprintf("frikky/shuffle:%s", baseTag))
possibleNames = append(possibleNames, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", baseTag))
if (arrayContains(possibleNames, image.Image)) {
if arrayContains(possibleNames, image.Image) {
log.Printf("[DEBUG] Image %s already downloaded that has been requested to download", image.Image)
resp.WriteHeader(200)
resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "image already present"}`)))
@@ -3293,7 +3578,7 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) {
}
log.Printf("[INFO] Downloading image %s", image.Image)
downloadDockerImageBackend(&http.Client{Timeout: 60 * time.Second}, image.Image)
downloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, image.Image)
// return success
resp.WriteHeader(200)
@@ -3326,6 +3611,7 @@ func runWebserver(listener net.Listener) {
//log.Fatal(http.Serve(listener, nil))
log.Printf("[DEBUG] NEW webserver setup")
http.Handle("/", r)