added cleanup for apps and worker for k8s

functions/k8s-confs/env-configmap.yaml

@@ -53,7 +53,7 @@ data:
   SSO_REDIRECT_URL: ""
   TZ: "Europe/Amsterdam \t\t\t\t\t"
   IS_KUBERNETES: "true"
-  REGISTRY_URL: "172.17.14.71:5000"
+  REGISTRY_URL: "shuffle-registry:5000"
   #REGISTRY_AUTH: ""
 kind: ConfigMap
 metadata:

functions/k8s-confs/orborus-deployment.yaml

@@ -24,10 +24,10 @@ spec:
         io.kompose.network/shuffle: "true"
         io.kompose.service: orborus
     spec:
-      volumes:
+      # volumes:
-        - name: docker-socket
+      #   - name: docker-socket
-          hostPath:
+      #     hostPath:
-            path: /var/run/docker.sock
+      #       path: /var/run/docker.sock
       containers:
         - env:
             - name: BASE_URL
@@ -61,9 +61,9 @@ spec:
           imagePullPolicy: Never
           name: shuffle-orborus
           resources: {}
-          volumeMounts:
+          # volumeMounts:
-            - name: docker-socket 
+          #   - name: docker-socket 
-              mountPath: /var/run/docker.sock
+          #     mountPath: /var/run/docker.sock
       hostname: shuffle-orborus
       restartPolicy: Always
 status: {}

functions/k8s-confs/shuffle-registry.yaml

@@ -0,0 +1,68 @@
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: shuffle-registry
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: shuffle-registry
+  template:
+    metadata:
+      labels:
+        app: shuffle-registry
+    spec:
+      initContainers:
+      - name: generate-certs
+        image: frapsoft/openssl
+        command: ["openssl"]
+        args:
+        - req
+        - -newkey
+        - rsa:4096
+        - -nodes
+        - -sha256
+        - -x509
+        - -days
+        - "365"
+        - -keyout
+        - /certs/domain.key
+        - -out
+        - /certs/domain.crt
+        - -subj
+        - "/CN=shuffle-registry.default.svc.cluster.local"
+        volumeMounts:
+        - name: certs
+          mountPath: /certs
+      containers:
+      - name: registry
+        image: registry:2
+        ports:
+        - containerPort: 5000
+        env:
+        - name: REGISTRY_HTTP_TLS_CERTIFICATE
+          value: "/certs/domain.crt"
+        - name: REGISTRY_HTTP_TLS_KEY
+          value: "/certs/domain.key"
+        volumeMounts:
+        - name: certs
+          mountPath: "/certs"
+      volumes:
+      - name: certs
+        emptyDir: {}
+
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: shuffle-registry
+spec:
+  selector:
+    app: shuffle-registry
+  ports:
+    - protocol: TCP
+      port: 5000
+      targetPort: 5000

functions/onprem/orborus/orborus.go

@@ -612,6 +612,7 @@ func deployWorker(image string, identifier string, env []string, executionReques
 				Labels: map[string]string{"app": "shuffle-worker"},
 			},
 			Spec: corev1.PodSpec{
+				RestartPolicy: "Never",
 				NodeSelector: map[string]string{
 					"node": "master",
 				},
@@ -1306,58 +1307,86 @@ func main() {
 // Is this ok to do with Docker? idk :)
 func getRunningWorkers(ctx context.Context, workerTimeout int) int {
 	//log.Printf("[DEBUG] Getting running workers with API version %s", dockerApiVersion)
-	containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{
+	counter := 0
-		All: true,
+	if os.Getenv("IS_KUBERNETES") == "true" {
-	})
+		log.Printf("[INFO] getting running workers in kubernetes")
 
-	// Automatically updates the version
+		thresholdTime := time.Now().Add(time.Duration(-workerTimeout) * time.Second)
-	if err != nil {
-		log.Printf("[ERROR] Error getting containers: %s", err)
 
-		newVersionSplit := strings.Split(fmt.Sprintf("%s", err), "version is")
+		clientset, err := getKubernetesClient()
-		if len(newVersionSplit) > 1 {
+		if err != nil {
-			//dockerApiVersion = strings.TrimSpace(newVersionSplit[1])
+			log.Printf("[ERROR] Failed getting kubernetes client: %s", err)
-			log.Printf("[DEBUG] WANT to change the API version to default to %s?", strings.TrimSpace(newVersionSplit[1]))
+			return 0
 		}
 
-		return maxConcurrency
+		labelSelector := "app=shuffle-worker"
-	}
+		pods, podErr := clientset.CoreV1().Pods("shuffle").List(ctx, metav1.ListOptions{
+			LabelSelector: labelSelector,
+		})
+		if podErr != nil {
+			log.Printf("[ERROR] Failed getting running workers: %s", podErr)
+			return 0
+		}
 
-	currenttime := time.Now().Unix()
+		for _, pod := range pods.Items {
-	counter := 0
+			if pod.Status.Phase == "Running" && pod.CreationTimestamp.Time.After(thresholdTime) {
-	for _, container := range containers {
+				counter++
-		// Skip random containers. Only handle things related to Shuffle.
+			}
-		if !strings.Contains(container.Image, baseimagename) {
+		}
-			shuffleFound := false
+	} else {
-			for _, item := range container.Labels {
+
-				if item == "shuffle" {
+		containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{
-					shuffleFound = true
+			All: true,
+		})
+	
+		// Automatically updates the version
+		if err != nil {
+			log.Printf("[ERROR] Error getting containers: %s", err)
+	
+			newVersionSplit := strings.Split(fmt.Sprintf("%s", err), "version is")
+			if len(newVersionSplit) > 1 {
+				//dockerApiVersion = strings.TrimSpace(newVersionSplit[1])
+				log.Printf("[DEBUG] WANT to change the API version to default to %s?", strings.TrimSpace(newVersionSplit[1]))
+			}
+	
+			return maxConcurrency
+		}
+	
+		currenttime := time.Now().Unix()
+
+		for _, container := range containers {
+			// Skip random containers. Only handle things related to Shuffle.
+			if !strings.Contains(container.Image, baseimagename) {
+				shuffleFound := false
+				for _, item := range container.Labels {
+					if item == "shuffle" {
+						shuffleFound = true
+						break
+					}
+				}
+	
+				// Check image name
+				if !shuffleFound {
+					continue
+				}
+				//} else {
+				//	log.Printf("NAME: %s", container.Image)
+			}
+	
+			for _, name := range container.Names {
+				// FIXME - add name_version_uid_uid regex check as well
+				if !strings.HasPrefix(name, "/worker") {
+					continue
+				}
+	
+				//log.Printf("Time: %d - %d", currenttime-container.Created, int64(workerTimeout))
+				if container.State == "running" && currenttime-container.Created < int64(workerTimeout) {
+					counter += 1
 					break
 				}
 			}
-
-			// Check image name
-			if !shuffleFound {
-				continue
-			}
-			//} else {
-			//	log.Printf("NAME: %s", container.Image)
-		}
-
-		for _, name := range container.Names {
-			// FIXME - add name_version_uid_uid regex check as well
-			if !strings.HasPrefix(name, "/worker") {
-				continue
-			}
-
-			//log.Printf("Time: %d - %d", currenttime-container.Created, int64(workerTimeout))
-			if container.State == "running" && currenttime-container.Created < int64(workerTimeout) {
-				counter += 1
-				break
-			}
 		}
 	}
-
 	return counter
 }
 

functions/onprem/worker/worker.go

@@ -112,6 +112,18 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
 	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)
 	//reason := "Error in execution"
 
+	// if os.Getenv("IS_KUBERNETES") == "true" {
+	// 	log.Printf("[DEBUG][%s] Running in Kubernetes shutting down")
+	// 	workerPod := fmt.Sprintf("worker-%s", workflowExecution.ExecutionId)
+	// 	namespace := "shuffle"
+	// 	clientset, err := getKubernetesClient()
+	// 	if err != nil {
+	// 		log.Printf("[ERROR] Error getting kubernetes client: %s", err)
+	// 		return
+	// 	}
+
+	// } else {
+
 	sleepDuration := 1
 	if handleResultSend && requestsSent < 2 {
 		shutdownData, err := json.Marshal(workflowExecution)
@@ -181,6 +193,8 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason
 	os.Exit(3)
 }
 
+// }
+
 func isRunningInCluster() bool {
 	_, existsHost := os.LookupEnv("KUBERNETES_SERVICE_HOST")
 	_, existsPort := os.LookupEnv("KUBERNETES_SERVICE_PORT")
@@ -225,214 +239,97 @@ func getKubernetesClient() (*kubernetes.Clientset, error) {
 func deployApp(cli *dockerclient.Client, image string, identifier string, env []string, workflowExecution shuffle.WorkflowExecution, action shuffle.Action) error {
 	log.Printf("HELLO FROM DEPLOYAPP")
 	log.Printf("image: %s", image)
+	log.Printf("identifier: %s", identifier)
 
 	log.Printf("IS_KUBERNETES: %s", os.Getenv("IS_KUBERNETES"))
 	log.Printf("REGISTRY_NAME: %s", os.Getenv("REGISTRY_NAME"))
 
-	namespace := "shuffle"
+	if os.Getenv("IS_KUBERNETES") == "true" {
 
-	envMap := make(map[string]string)
+		namespace := "shuffle"
-	for _, envStr := range env {
+
-		parts := strings.SplitN(envStr, "=", 2)
+		envMap := make(map[string]string)
-		if len(parts) == 2 {
+		for _, envStr := range env {
-			envMap[parts[0]] = parts[1]
+			parts := strings.SplitN(envStr, "=", 2)
+			if len(parts) == 2 {
+				envMap[parts[0]] = parts[1]
+			}
 		}
-	}
 
-	clientset, err := getKubernetesClient()
+		clientset, err := getKubernetesClient()
-	if err != nil {
+		if err != nil {
-		fmt.Println("[ERROR]Error getting kubernetes client:", err)
+			fmt.Println("[ERROR]Error getting kubernetes client:", err)
-		os.Exit(1)
+			os.Exit(1)
-	}
+		}
 
-	log.Printf("[DEBUG] Got kubernetes client")
+		log.Printf("[DEBUG] Got kubernetes client")
-	str := strings.ToLower(identifier)
+		str := strings.ToLower(identifier)
-	strSplit := strings.Split(str, "_")
+		strSplit := strings.Split(str, "_")
-	value := strSplit[0]
+		value := strSplit[0]
-	value = strings.ReplaceAll(value, "_", "-")
+		value = strings.ReplaceAll(value, "_", "-")
 
-	//fix naming convention
+		//fix naming convention
-	podUuid := uuid.NewV4().String()
+		podUuid := uuid.NewV4().String()
-	podName := fmt.Sprintf("%s-%s", value, podUuid)
+		podName := fmt.Sprintf("%s-%s", value, podUuid)
 
-	pod := &corev1.Pod{
+		pod := &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
+			ObjectMeta: metav1.ObjectMeta{
-			Name: podName,
+				Name: podName,
-			Labels: map[string]string{
+				Labels: map[string]string{
-				"app": "shuffle-app",
+					"app":         "shuffle-app",
-			},
+					"executionId": workflowExecution.ExecutionId,
-		},
+				},
-		Spec: corev1.PodSpec{
+			},
-			Containers: []corev1.Container{
+			Spec: corev1.PodSpec{
-				{
+				RestartPolicy: "Never",
-					Name:  value,
+				Containers: []corev1.Container{
-					Image: image,
+					{
-					Env:   buildEnvVars(envMap),
+						Name:  value,
+						Image: image,
+						Env:   buildEnvVars(envMap),
+					},
 				},
 			},
-		},
-	}
-
-	createdPod, err := clientset.CoreV1().Pods(namespace).Create(context.Background(), pod, metav1.CreateOptions{})
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "Error creating pod: %v\n", err)
-		os.Exit(1)
-	}
-
-	fmt.Printf("Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace)
-
-	// workerPod := fmt.Sprintf("worker-%s", workflowExecution.ExecutionId)
-	// pod, err := clientset.CoreV1().Pods(namespace).Get(context.TODO(), workerPod, metav1.GetOptions{})
-	// if err != nil {
-	// 	fmt.Printf("Error getting pod: %v\n", err)
-	// 	os.Exit(1)
-	// }
-	// log.Printf("[DEBUG] Got pod %s", pod.Name)
-
-	// container := corev1.EphemeralContainer{
-	// 	EphemeralContainerCommon: corev1.EphemeralContainerCommon{
-	// 	Image:           image,
-	//     Env:             buildEnvVars(envMap),
-	//     ImagePullPolicy: corev1.PullIfNotPresent,
-	// 	},
-	// }
-
-	// pod.Spec.EphemeralContainers = append(pod.Spec.EphemeralContainers, container)
-
-	// err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
-	//     _, err := clientset.CoreV1().Pods(namespace).Update(context.Background(), pod, metav1.UpdateOptions{})
-	//     return err
-	// })
-	// if err != nil {
-	//     log.Fatalf("Failed to update Pod %v", err)
-	// }
-
-	// fmt.Printf("Ephemeral container added to Pod \n")
-
-	// ephemeralContainer := corev1.EphemeralContainer{
-	// 	EphemeralContainerCommon: corev1.EphemeralContainerCommon{
-	// 		Image: image,
-	// 		Env:   buildEnvVars(envMap),
-	// 	},
-	// }
-
-	// patchBytes := []byte(fmt.Sprintf(`[
-	// 	{
-	// 		"op": "add",
-	// 		"path": "/spec/ephemeralContainers",
-	// 		"value": %s
-	// 	}
-	// ]`, ephemeralContainer))
-
-	// _, err = clientset.CoreV1().Pods(namespace).PatchEphemeral(context.TODO(), pod.Name, containerName, patchBytes, metav1.PatchOptions{})
-	// if err != nil {
-	// 	fmt.Printf("Error adding ephemeral container: %v\n", err)
-	// 	os.Exit(1)
-	// }
-
-	// fmt.Println("Ephemeral container added successfully")
-
-	if action.AppName == "shuffle-subflow" {
-		// Automatic replacement of URL
-		for paramIndex, param := range action.Parameters {
-			if param.Name != "backend_url" {
-				continue
-			}
-
-			if strings.Contains(param.Value, "shuffle-backend") {
-				// Automatic replacement as this is default
-				action.Parameters[paramIndex].Value = os.Getenv("BASE_URL")
-				log.Printf("[DEBUG][%s] Replaced backend_url with %s", workflowExecution.ExecutionId, os.Getenv("BASE_URL"))
-			}
 		}
+
+		createdPod, err := clientset.CoreV1().Pods(namespace).Create(context.Background(), pod, metav1.CreateOptions{})
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "Error creating pod: %v\n", err)
+			os.Exit(1)
+		}
+		fmt.Printf("Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace)
+	} else {
+		// docker part
 	}
 
-	// Max 10% CPU every second
+	return nil
-	//CPUShares: 128,
+}
-	//CPUQuota:  10000,
-	//CPUPeriod: 100000,
 
-	// hostConfig := &container.HostConfig{
+func cleanupExecution(clientset *kubernetes.Clientset, workflowExecution shuffle.WorkflowExecution, namespace string) error {
-	// 	LogConfig: container.LogConfig{
-	// 		Type: "json-file",
-	// 		Config: map[string]string{
-	// 			"max-size": "10m",
-	// 		},
-	// 	},
-	// 	Resources: container.Resources{},
-	// }
 
-	// hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:worker-%s", workflowExecution.ExecutionId))
+	workerName := fmt.Sprintf("worker-%s", workflowExecution.ExecutionId)
+	labelSelector := fmt.Sprintf("app=shuffle-app,executionId=%s", workflowExecution.ExecutionId)
 
-	// Removing because log extraction should happen first
+	podList, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
-	// if cleanupEnv == "true" {
+		LabelSelector: labelSelector,
-	// 	hostConfig.AutoRemove = true
+	})
-	// }
+	if err != nil {
+		return fmt.Errorf("[ERROR]failed to list apps with label selector %s: %v", labelSelector, err)
+	}
 
-	// FIXME: Add proper foldermounts here
+	for _, pod := range podList.Items {
-	//log.Printf("\n\nPRE FOLDERMOUNT\n\n")
+		err := clientset.CoreV1().Pods(namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{})
-	//volumeBinds := []string{"/tmp/shuffle-mount:/rules"}
+		if err != nil {
-	//volumeBinds := []string{"/tmp/shuffle-mount:/rules"}
+			return fmt.Errorf("failed to delete app %s: %v", pod.Name, err)
-	// volumeBinds := []string{}
+		}
-	// if len(volumeBinds) > 0 {
+		fmt.Printf("App %s in namespace %s deleted.\n", pod.Name, namespace)
-	// 	log.Printf("[DEBUG] Setting up binds for container!")
+	}
-	// 	hostConfig.Binds = volumeBinds
-	// 	hostConfig.Mounts = []mount.Mount{}
-	// 	for _, bind := range volumeBinds {
-	// 		if !strings.Contains(bind, ":") || strings.Contains(bind, "..") || strings.HasPrefix(bind, "~") {
-	// 			log.Printf("[WARNING] Bind %s is invalid.", bind)
-	// 			continue
-	// 		}
-
-	// 		log.Printf("[DEBUG] Appending bind %s", bind)
-	// 		bindSplit := strings.Split(bind, ":")
-	// 		sourceFolder := bindSplit[0]
-	// 		destinationFolder := bindSplit[0]
-	// 		hostConfig.Mounts = append(hostConfig.Mounts, mount.Mount{
-	// 			Type:   mount.TypeBind,
-	// 			Source: sourceFolder,
-	// 			Target: destinationFolder,
-	// 		})
-	// 	}
-	// } else {
-	// 	//log.Printf("[WARNING] Not mounting folders")
-	// }
-
-	// config := &container.Config{
-	// 	Image: image,
-	// 	Env:   env,
-	// }
-
-	// Checking as late as possible, just in case.
-	// newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID)
-	// _, err := shuffle.GetCache(ctx, newExecId)
-	// if err == nil {
-	// 	log.Printf("\n\n[DEBUG] Result for %s already found - returning\n\n", newExecId)
-	// 	return 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 {
-	// 	log.Printf("[DEBUG] Adding %s to cache. Name: %s", newExecId, action.Name)
-	// }
-
-	// if action.ExecutionDelay > 0 {
-	// 	log.Printf("[DEBUG] Running app %s in docker with delay of %d", action.Name, action.ExecutionDelay)
-	// 	waitTime := time.Duration(action.ExecutionDelay) * time.Second
-
-	// 	time.AfterFunc(waitTime, func() {
-	// 		DeployContainer(ctx, cli, config, hostConfig, identifier, workflowExecution, newExecId)
-	// 	})
-	// } else {
-	// 	log.Printf("[DEBUG] Running app %s in docker NORMALLY as there is no delay set with identifier %s", action.Name, identifier)
-	// 	returnvalue := DeployContainer(ctx, cli, config, hostConfig, identifier, workflowExecution, newExecId)
-	// 	log.Printf("[DEBUG] Normal deploy ret: %s", returnvalue)
-	// 	return returnvalue
-	// }
 
+	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)
+	}
+	fmt.Printf("[DEBUG]  %s in namespace %s deleted.\n", workerName, namespace)
 	return nil
 }
 
@@ -1026,7 +923,17 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) {
 			log.Printf("[INFO][%s] BREAKING BECAUSE RESULTS IS SAME LENGTH AS ACTIONS. SHOULD CHECK ALL RESULTS FOR WHETHER THEY'RE DONE", workflowExecution.ExecutionId)
 			validateFinished(workflowExecution)
 			log.Printf("[DEBUG][%s] Shutting down (17)", workflowExecution.ExecutionId)
-			shutdown(workflowExecution, "", "", true)
+			if os.Getenv("IS_KUBERNETES") == "true" {
+				// log.Printf("workflow execution: %#v", workflowExecution)
+				clientset, err := getKubernetesClient()
+				if err != nil {
+					fmt.Println("[ERROR]Error getting kubernetes client:", err)
+					os.Exit(1)
+				}
+				cleanupExecution(clientset, workflowExecution, "shuffle")
+			} else {
+				shutdown(workflowExecution, "", "", true)
+			}
 			return
 		}
 	}
@@ -1243,7 +1150,18 @@ func handleDefaultExecution(client *http.Client, req *http.Request, workflowExec
 		if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" {
 			log.Printf("[INFO][%s] Workflow execution is finished. Exiting worker.", workflowExecution.ExecutionId)
 			log.Printf("[DEBUG] Shutting down (20)")
-			shutdown(workflowExecution, "", "", true)
+			//handle workerssssssssss
+			if os.Getenv("IS_KUBERNETES") == "true" {
+				// log.Printf("workflow execution: %#v", workflowExecution)
+				clientset, err := getKubernetesClient()
+				if err != nil {
+					fmt.Println("[ERROR]Error getting kubernetes client:", err)
+					os.Exit(1)
+				}
+				cleanupExecution(clientset, workflowExecution, "shuffle")
+			} else {
+				shutdown(workflowExecution, "", "", true)
+			}
 		}
 
 		log.Printf("[INFO][%s] Status: %s, Results: %d, actions: %d", workflowExecution.ExecutionId, workflowExecution.Status, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)+extra)