diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index f9bfdc7f..d11b6e37 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -1,6 +1,6 @@ module shuffle-shared -//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared +replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared go 1.19 @@ -53,7 +53,7 @@ require ( github.com/docker/go-connections v0.4.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/emirpasic/gods v1.18.1 // indirect - github.com/frikky/schemaless v0.0.8 // indirect + github.com/frikky/schemaless v0.0.9 // indirect github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect github.com/go-logr/logr v1.2.4 // indirect github.com/go-openapi/jsonpointer v0.19.5 // indirect diff --git a/backend/go-app/go.sum b/backend/go-app/go.sum index f1acfbbb..c12f4bd9 100644 --- a/backend/go-app/go.sum +++ b/backend/go-app/go.sum @@ -178,8 +178,8 @@ github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM github.com/frikky/kin-openapi v0.41.0/go.mod h1:ev9OZAw7Bv5p0w93j91++6a1ElPzGcCofst+kmrWsj4= github.com/frikky/kin-openapi v0.42.0 h1:d5Z6vnuQ6RnCCPIxZaDL+TH2ODLxT8abytOt+Zh+Kd0= github.com/frikky/kin-openapi v0.42.0/go.mod h1:ev9OZAw7Bv5p0w93j91++6a1ElPzGcCofst+kmrWsj4= -github.com/frikky/schemaless v0.0.8 h1:9ekdXCgVSKb18g4e6/Jr0C3d5L1gfpYhJoJwQOOMMJo= -github.com/frikky/schemaless v0.0.8/go.mod h1:mooDxY+D6weHjhKvjy3+IE9S7P4g4cpNnidkdRv/cHQ= +github.com/frikky/schemaless v0.0.9 h1:RzNLPkJq5c4nlm5iLiTndFcbeQxdMGJIj266wSGt2+8= +github.com/frikky/schemaless v0.0.9/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/fsouza/go-dockerclient v1.9.7 h1:FlIrT71E62zwKgRvCvWGdxRD+a/pIy+miY/n3MXgfuw= @@ -457,8 +457,6 @@ github.com/seccomp/libseccomp-golang v0.9.2-0.20220502022130-f33da4d89646/go.mod github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo= 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.14 h1:ZO6Stk5d+ZvH+cDNOJ844ex0EXmzLf0iMVep/NSj/g8= -github.com/shuffle/shuffle-shared v0.6.14/go.mod h1:fsWCs0nsCS/3PN9BHl6qPN0PrTctT/TWUMH9ElTsKTg= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 7e01b46c..042fc6d6 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -74,6 +74,7 @@ var newWorkerImage = os.Getenv("SHUFFLE_WORKER_IMAGE") var dockerSwarmBridgeMTU = os.Getenv("SHUFFLE_SWARM_BRIDGE_DEFAULT_MTU") var dockerSwarmBridgeInterface = os.Getenv("SHUFFLE_SWARM_BRIDGE_DEFAULT_INTERFACE") var isKubernetes = os.Getenv("IS_KUBERNETES") +var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") var maxCPUPercent = 95 // var baseimagename = "docker.pkg.github.com/shuffle/shuffle" @@ -103,6 +104,7 @@ var orborusLabel = os.Getenv("SHUFFLE_ORBORUS_LABEL") var memcached = os.Getenv("SHUFFLE_MEMCACHED") var executionIds = []string{} +var namespacemade = false // For K8s var dockercli *dockerclient.Client var containerId string @@ -663,15 +665,52 @@ func handleBackendImageDownload(ctx context.Context, images 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"))) + } if isKubernetes == "true" { - if len(os.Getenv("REGISTRY_URL")) > 0 && os.Getenv("REGISTRY_URL") != "" { - env = append(env, fmt.Sprintf("REGISTRY_URL=%s", os.Getenv("REGISTRY_URL"))) - env = append(env, fmt.Sprintf("IS_KUBERNETES=%s", os.Getenv("IS_KUBERNETES"))) + env = append(env, fmt.Sprintf("IS_KUBERNETES=%s", os.Getenv("IS_KUBERNETES"))) + env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE"))) + + clientset, err := getKubernetesClient() + if err != nil { + log.Printf("[ERROR] Error getting kubernetes client:", err) + return err } - image = os.Getenv("SHUFFLE_KUBERNETES_WORKER") - log.Printf("[DEBUG] using worker image:", image) + // Check if namespace exist as variable. If so, make it + if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 && !namespacemade { + kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") + + // Make the namespace + namespace := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: os.Getenv("KUBERNETES_NAMESPACE"), + }, + } + + _, err := clientset.CoreV1().Namespaces().Create(context.Background(), namespace, metav1.CreateOptions{}) + if err != nil { + if !strings.Contains(strings.ToLower(fmt.Sprintf("%s", err)), "already exists") { + log.Printf("[ERROR] Failed creating Kubernetes namespace: %s", err) + } else { + namespacemade = true + } + } else { + namespacemade = true + } + } + + if len(kubernetesNamespace) == 0 { + kubernetesNamespace = "default" + } + + kubernetesImage := os.Getenv("SHUFFLE_KUBERNETES_WORKER") + if len(kubernetesImage) == 0 { + kubernetesImage = image + } + log.Printf("[DEBUG] Using Kubernetes worker image '%s'", kubernetesImage) // image = "shuffle-worker:v1" //hard coded image name to test locally envMap := make(map[string]string) @@ -682,12 +721,8 @@ func deployWorker(image string, identifier string, env []string, executionReques } } - clientset, err := getKubernetesClient() - if err != nil { - log.Printf("[ERROR] Error getting kubernetes client:", err) - return err - } - + // While testing: + // kubectl delete pods --all --all-namespaces; kubectl delete services --all --all-namespaces pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: identifier, @@ -695,27 +730,31 @@ func deployWorker(image string, identifier string, env []string, executionReques }, Spec: corev1.PodSpec{ RestartPolicy: "Never", - // once images is pushed, we can remove this - // keep this when running locally + DNSPolicy: "Default", // NodeSelector: map[string]string{ // "node": "master", // }, Containers: []corev1.Container{ { Name: identifier, - Image: image, + Image: kubernetesImage, Env: buildEnvVars(envMap), + + //ImagePullPolicy: "Never", + ImagePullPolicy: corev1.PullIfNotPresent, }, }, }, } + // Check if running on ARM or x86 to download the correct image + // Add environment variables // pod.Spec.Containers[0].Env = buildEnvVars(envMap) - createdPod, err := clientset.CoreV1().Pods("shuffle").Create(context.Background(), pod, metav1.CreateOptions{}) + createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{}) if err != nil { - log.Printf("[ERROR] Failed creating pod: %v", err) + //log.Printf("[ERROR] Failed creating pod: %v", err) return err } @@ -1773,6 +1812,10 @@ func main() { executionIds = append(executionIds, execution.ExecutionId) } else { log.Printf("[WARNING] Execution ID '%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) + } } } @@ -1985,7 +2028,7 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int { //log.Printf("[DEBUG] Getting running workers with API version %s", dockerApiVersion) counter := 0 if isKubernetes == "true" { - log.Printf("[INFO] getting running workers in kubernetes") + log.Printf("[INFO] Getting running workers in kubernetes") thresholdTime := time.Now().Add(time.Duration(-workerTimeout) * time.Second) @@ -1996,7 +2039,7 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int { } labelSelector := "app=shuffle-worker" - pods, podErr := clientset.CoreV1().Pods("shuffle").List(ctx, metav1.ListOptions{ + pods, podErr := clientset.CoreV1().Pods(kubernetesNamespace).List(ctx, metav1.ListOptions{ LabelSelector: labelSelector, }) if podErr != nil { @@ -2009,6 +2052,10 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int { counter++ } } + + if counter > 0 { + log.Printf("[INFO] Found %d running workers in Orborus", counter) + } } else { containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{ diff --git a/functions/onprem/worker/go.mod b/functions/onprem/worker/go.mod index b51d219d..bae8288d 100644 --- a/functions/onprem/worker/go.mod +++ b/functions/onprem/worker/go.mod @@ -8,7 +8,7 @@ require ( github.com/docker/docker v23.0.3+incompatible github.com/gorilla/mux v1.8.0 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.5.98 + github.com/shuffle/shuffle-shared v0.6.16 k8s.io/api v0.28.3 k8s.io/apimachinery v0.28.3 k8s.io/client-go v0.28.3 @@ -38,7 +38,7 @@ require ( github.com/emicklei/go-restful/v3 v3.9.0 // indirect github.com/emirpasic/gods v1.18.1 // indirect github.com/frikky/kin-openapi v0.41.0 // indirect - github.com/frikky/schemaless v0.0.6 // indirect + github.com/frikky/schemaless v0.0.9 // 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 diff --git a/functions/onprem/worker/go.sum b/functions/onprem/worker/go.sum index f2520ce2..3b7752e4 100644 --- a/functions/onprem/worker/go.sum +++ b/functions/onprem/worker/go.sum @@ -89,6 +89,7 @@ github.com/bradfitz/slice v0.0.0-20180809154707-2b758aa73013 h1:/P9/RL0xgWE+ehnC github.com/bradfitz/slice v0.0.0-20180809154707-2b758aa73013/go.mod h1:pccXHIvs3TV/TUqSNyEvF99sxjX2r4FFRIyw6TZY9+w= github.com/bwesterb/go-ristretto v1.2.3/go.mod h1:fUIoIZaG73pV5biE2Blr2xEzDoMj7NFEuV9ekS419A0= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= +github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI= github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI= github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU= @@ -97,6 +98,8 @@ github.com/cloudflare/circl v1.3.3 h1:fE/Qz0QdIGqeWfnwq0RE0R7MI51s0M2E4Ga9kq5AEM github.com/cloudflare/circl v1.3.3/go.mod h1:5XYMA4rFBvNIrhs50XuiBJ15vF2pZn4nnUKZrLbUZFA= 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/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI= +github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/cyphar/filepath-securejoin v0.2.4 h1:Ugdm7cg7i6ZK6x3xDF1oEu1nfkyfH53EtKeQYTC3kyg= github.com/cyphar/filepath-securejoin v0.2.4/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4= @@ -122,11 +125,14 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po= +github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= github.com/frikky/kin-openapi v0.41.0 h1:oMmjo+ekGS971lb3KLeZZOqRDZOwWi3+g/OiSWP08+s= github.com/frikky/kin-openapi v0.41.0/go.mod h1:ev9OZAw7Bv5p0w93j91++6a1ElPzGcCofst+kmrWsj4= github.com/frikky/schemaless v0.0.6 h1:mPWbqCxiOz0HUmdN+IiVOHqquCzA0aachzOdMTCaKtg= github.com/frikky/schemaless v0.0.6/go.mod h1:mooDxY+D6weHjhKvjy3+IE9S7P4g4cpNnidkdRv/cHQ= +github.com/frikky/schemaless v0.0.9 h1:RzNLPkJq5c4nlm5iLiTndFcbeQxdMGJIj266wSGt2+8= +github.com/frikky/schemaless v0.0.9/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= @@ -224,6 +230,7 @@ github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXi github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0= github.com/google/martian/v3 v3.1.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0= github.com/google/martian/v3 v3.2.1 h1:d8MncMlErDFTwQGBK1xhv026j9kqhvw1Qv9IbWT1VLQ= +github.com/google/martian/v3 v3.2.1/go.mod h1:oBOf6HBosgwRXnUGWUB05QECsc6uvmMiJ3+6W4l/CUk= github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc= github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc= github.com/google/pprof v0.0.0-20191218002539-d4f498aebedc/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM= @@ -236,6 +243,7 @@ github.com/google/pprof v0.0.0-20201023163331-3e6fc7fc9c4c/go.mod h1:kpwsk12EmLe github.com/google/pprof v0.0.0-20201218002935-b9804c9f04c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE= github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE= github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJYCmNdQXq6neHJOYx3V6jnqNEec= +github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE= github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= @@ -365,6 +373,8 @@ 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.5.86 h1:ZHQgZ4siSWgi5gttxeSMdjsaH9SbGTzrA/GO6aICO2U= github.com/shuffle/shuffle-shared v0.5.86/go.mod h1:Lg6/+qjQlWzNKwj4/4ATpvScyP2JQGLkTPlNlRM6RJk= +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/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/skeema/knownhosts v1.2.1 h1:SHWdIUa82uGZz+F+47k8SY4QhhI291cXCpopT1lK2AQ= diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 58cb4de9..e47059f6 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -53,6 +53,7 @@ var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME") var dockerApiVersion = strings.ToLower(os.Getenv("DOCKER_API_VERSION")) var baseimagename = "frikky/shuffle" +var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") // var baseimagename = os.Getenv("SHUFFLE_BASE_IMAGE_NAME") @@ -383,8 +384,11 @@ 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" { - namespace := "shuffle" - localRegistry := os.Getenv("REGISTRY_URL") + if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 { + kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") + } else { + kubernetesNamespace = "default" + } envMap := make(map[string]string) for _, envStr := range env { @@ -397,7 +401,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] clientset, err := getKubernetesClient() if err != nil { log.Printf("[ERROR] Failed getting kubernetes: %s [INFO] Setting kubernetes to false to enable running Shuffle with Docker for the next iterations.", err) - isKubernetes = "false" return err } @@ -406,7 +409,8 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] value := strSplit[0] value = strings.ReplaceAll(value, "_", "-") - // checking if app is generated or not + // Checking if app is generated or not + localRegistry := os.Getenv("REGISTRY_URL") /* appDetails := strings.Split(image, ":")[1] appDetailsSplit := strings.Split(appDetails, "_") @@ -440,7 +444,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] } } - log.Printf("[DEBUG] Got kubernetes client to run image '%s'", image) + log.Printf("[DEBUG] Got kubernetes with namespace %#v to run image '%s'", kubernetesNamespace, image) //fix naming convention podUuid := uuid.NewV4().String() @@ -455,25 +459,30 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] }, }, Spec: corev1.PodSpec{ + RestartPolicy: "Never", // As a crash is not useful in this context + DNSPolicy: "Default", // NodeName: "worker1" - RestartPolicy: "Never", Containers: []corev1.Container{ { Name: value, Image: image, Env: buildEnvVars(envMap), - // ImagePullPolicy: corev1.PullAlways, + + // Pull if not available + ImagePullPolicy: corev1.PullIfNotPresent, }, }, }, } - createdPod, err := clientset.CoreV1().Pods(namespace).Create(context.Background(), pod, metav1.CreateOptions{}) + createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{}) if err != nil { - fmt.Fprintf(os.Stderr, "Error creating pod: %v", err) + 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("[DEBUG] Created pod %q in namespace %q", createdPod.Name, createdPod.Namespace) + return nil } @@ -612,7 +621,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] return nil } -func cleanupExecution(clientset *kubernetes.Clientset, workflowExecution shuffle.WorkflowExecution, namespace string) error { +func cleanupKubernetesExecution(clientset *kubernetes.Clientset, workflowExecution shuffle.WorkflowExecution, namespace string) error { workerName := fmt.Sprintf("worker-%s", workflowExecution.ExecutionId) labelSelector := fmt.Sprintf("app=shuffle-app,executionId=%s", workflowExecution.ExecutionId) @@ -1385,7 +1394,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { log.Println("[ERROR] Error getting kubernetes client (1):", err) os.Exit(1) } - cleanupExecution(clientset, workflowExecution, "shuffle") + cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace) } else { shutdown(workflowExecution, "", "", true) } @@ -1595,7 +1604,7 @@ func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.Workflow os.Exit(1) } - cleanupExecution(clientset, workflowExecution, "shuffle") + cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace) } else { shutdown(workflowExecution, "", "", true) } @@ -1691,7 +1700,7 @@ func handleDefaultExecutionWrapper(ctx context.Context, workflowExecution shuffl log.Println("[ERROR] Error getting kubernetes client (2):", err) os.Exit(1) } - cleanupExecution(clientset, workflowExecution, "shuffle") + cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace) } else { shutdown(workflowExecution, "", "", true) } @@ -1708,7 +1717,8 @@ func handleDefaultExecutionWrapper(ctx context.Context, workflowExecution shuffl log.Println("[ERROR] Error getting kubernetes client (3):", err) os.Exit(1) } - cleanupExecution(clientset, workflowExecution, "shuffle") + + cleanupKubernetesExecution(clientset, workflowExecution, kubernetesNamespace) } else { shutdown(workflowExecution, "", "", true) }