Merge branch 'nightly' of https://github.com/shuffle/shuffle into nightly

functions/onprem/orborus/orborus.go

@@ -217,8 +217,7 @@ func skipCheckInCleanup(name string) bool {
 func cleanupExistingNodes(ctx context.Context) error {
 
 	if isKubernetes == "true" {
-		// of course, this doesn't clean up "nodes" but
+		// Cleanup all workers created by orborus and all apps created by workers.
-		// rather pods, services, roles etc.
 
 		if kubernetesNamespace == "" {
 			kubernetesNamespace = "default"
@@ -230,62 +229,38 @@ func cleanupExistingNodes(ctx context.Context) error {
 			return err
 		}
 
-		// Delete all pods
-		pods, err := clientset.CoreV1().Pods(kubernetesNamespace).List(context.Background(), metav1.ListOptions{})
-		if err != nil {
-			log.Printf("[ERROR] Failed listing pods: %s", err)
-			return err
-		}
-
-		for _, pod := range pods.Items {
-			// check if pod.Name starts with:
-			// "backend-", "frontend-", "orborus-", "opensearch-" or "memcached-"
-			if skipCheckInCleanup(pod.Name) {
-				continue
-			}
-
-			err := clientset.CoreV1().Pods(kubernetesNamespace).Delete(context.Background(), pod.Name, metav1.DeleteOptions{})
-			if err != nil {
-				log.Printf("[ERROR] Failed deleting pod %s: %s", pod.Name, err)
-			}
-		}
-
 		// Delete all services
-		services, err := clientset.CoreV1().Services(kubernetesNamespace).List(context.Background(), metav1.ListOptions{})
+		services, err := clientset.CoreV1().Services(kubernetesNamespace).List(context.Background(), metav1.ListOptions{
+			LabelSelector: "app.kubernetes.io/name in (shuffle-worker, shuffle-app),app.kubernetes.io/managed-by in (shuffle-orborus, shuffle-worker)",
+		})
 		if err != nil {
 			log.Printf("[ERROR] Failed listing services: %s", err)
 			return err
 		}
 
 		for _, service := range services.Items {
-			if skipCheckInCleanup(service.Name) {
-				continue
-			}
-
 			err := clientset.CoreV1().Services(kubernetesNamespace).Delete(context.Background(), service.Name, metav1.DeleteOptions{})
 			if err != nil {
 				log.Printf("[ERROR] Failed deleting service %s: %s", service.Name, err)
 			}
 		}
 
-		deployments, err := clientset.AppsV1().Deployments(kubernetesNamespace).List(context.Background(), metav1.ListOptions{})
+		deployments, err := clientset.AppsV1().Deployments(kubernetesNamespace).List(context.Background(), metav1.ListOptions{
+			LabelSelector: "app.kubernetes.io/name in (shuffle-worker, shuffle-app),app.kubernetes.io/managed-by in (shuffle-orborus, shuffle-worker)",
+		})
 		if err != nil {
 			log.Printf("[ERROR] Failed listing deployments: %s", err)
 			return err
 		}
 
 		for _, deployment := range deployments.Items {
-			if skipCheckInCleanup(deployment.Name) {
-				continue
-			}
-
 			err := clientset.AppsV1().Deployments(kubernetesNamespace).Delete(context.Background(), deployment.Name, metav1.DeleteOptions{})
 			if err != nil {
 				log.Printf("[ERROR] Failed deleting deployment %s: %s", deployment.Name, err)
 			}
 		}
 
-		log.Printf("[INFO] Cleaned up all pods and services in namespace %s. Waiting 10 seconds for cleanup to reflect", kubernetesNamespace)
+		log.Printf("[INFO] Cleaned up all services and deployments in namespace %s. Waiting 10 seconds for cleanup to reflect", kubernetesNamespace)
 
 		time.Sleep(10 * time.Second)
 
@@ -1092,10 +1067,21 @@ func deployK8sWorker(image string, identifier string, env []string) error {
 		}
 	}
 
-	containerLabels := map[string]string{
+	labels := map[string]string{
+		"app.kubernetes.io/name":     "shuffle-worker",
+		"app.kubernetes.io/instance": identifier,
+		// "app.kubernetes.io/version":    "",
+		"app.kubernetes.io/part-of":    "shuffle",
+		"app.kubernetes.io/managed-by": "shuffle-orborus",
+		// Keep legacy labels for backward compatibility
 		"container": "shuffle-worker",
 	}
 
+	matchLabels := map[string]string{
+		"app.kubernetes.io/name":     "shuffle-worker",
+		"app.kubernetes.io/instance": identifier,
+	}
+
 	containerAttachment := corev1.Container{
 		Name:  identifier,
 		Image: kubernetesImage,
@@ -1198,16 +1184,17 @@ func deployK8sWorker(image string, identifier string, env []string) error {
 
 	deployment := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: identifier,
+			Name:   identifier,
+			Labels: labels,
 		},
 		Spec: appsv1.DeploymentSpec{
 			Replicas: &replicaNumberInt32,
 			Selector: &metav1.LabelSelector{
-				MatchLabels: containerLabels,
+				MatchLabels: matchLabels,
 			},
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
-					Labels: containerLabels,
+					Labels: labels,
 				},
 				Spec: corev1.PodSpec{
 					Containers: []corev1.Container{
@@ -1228,10 +1215,11 @@ func deployK8sWorker(image string, identifier string, env []string) error {
 	// kubectl expose deployment shuffle-workers --type=NodePort --port=33333 --target-port=33333
 	service := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: identifier,
+			Name:   identifier,
+			Labels: labels,
 		},
 		Spec: corev1.ServiceSpec{
-			Selector: containerLabels,
+			Selector: matchLabels,
 			Ports: []corev1.ServicePort{
 				{
 					Protocol:   "TCP",
@@ -2252,7 +2240,7 @@ func main() {
 				if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_START" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" {
 
 					os.Setenv("SHUFFLE_SKIP_PIPELINES", "false")
-					tenzirDisabled = false 
+					tenzirDisabled = false
 
 					// Running NEW or editing pipelines
 					err := handlePipeline(incRequest)
@@ -2274,7 +2262,7 @@ func main() {
 
 				} else if incRequest.Type == "CATEGORY_UPDATE" {
 					os.Setenv("SHUFFLE_SKIP_PIPELINES", "false")
-					tenzirDisabled = false 
+					tenzirDisabled = false
 
 					err = handleFileCategoryChange()
 					if err != nil {
@@ -2319,7 +2307,7 @@ func main() {
 
 					// Manual command = overrides to allow starting of Tenzir from the frontend anyway.
 					os.Setenv("SHUFFLE_SKIP_PIPELINES", "false")
-					tenzirDisabled = false 
+					tenzirDisabled = false
 
 					// Removed either way
 					toBeRemoved.Data = append(toBeRemoved.Data, incRequest)
@@ -2329,7 +2317,7 @@ func main() {
 						if strings.Contains(fmt.Sprintf("%s", err), "node available") {
 							// Disabling until UI is updated
 							os.Setenv("SHUFFLE_SKIP_PIPELINES", "true")
-							tenzirDisabled = true 
+							tenzirDisabled = true
 
 							log.Printf("[ERROR] Failed to start tenzir, reason: %s", err)
 							err = shuffle.CreateOrgNotification(
@@ -2648,7 +2636,7 @@ func deployTenzirNode() error {
 		// return errors.New("Pipelines are disabled by user with SHUFFLE_SKIP_PIPELINES")
 		//log.Printf("[INFO] Pipelines are enabled by user")
 	} else {
-		return errors.New("Pipelines are disabled by user with SHUFFLE_SKIP_PIPELINES") 
+		return errors.New("Pipelines are disabled by user with SHUFFLE_SKIP_PIPELINES")
 	}
 
 	if isKubernetes == "true" {
@@ -2848,12 +2836,12 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri
 		},
 	}
 
-	// FIXME: Is this necessary? Seems to screw up networking: 
+	// FIXME: Is this necessary? Seems to screw up networking:
 	// conflicting options: hostname and the network mode
 	/*
-	if isKubernetes != "true" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" {
+		if isKubernetes != "true" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" {
-		hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
+			hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId))
-	}
+		}
 	*/
 
 	resp, err := dockercli.ContainerCreate(ctx, config, hostConfig, networkingConfig, nil, containerName)
@@ -3061,7 +3049,6 @@ func createPipeline(command, identifier string) (string, error) {
 		return "", err
 	}
 
-
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
 		log.Printf("[ERROR] Failed reading response body: %s", err)
@@ -3079,8 +3066,8 @@ func createPipeline(command, identifier string) (string, error) {
 	}
 
 	type PipelineResponse struct {
-		ID string `json:"id"`
+		ID       string `json:"id"`
-		Message string `json:"message"`
+		Message  string `json:"message"`
 		Severity string `json:"severity"`
 	}
 
@@ -3303,7 +3290,7 @@ func handleFileCategoryChange() error {
 		tenzirStorageFolder = "/tmp/"
 	}
 
-	tenzirStorageFolder = strings.TrimRight(tenzirStorageFolder, "/") 
+	tenzirStorageFolder = strings.TrimRight(tenzirStorageFolder, "/")
 	sigmaPath := fmt.Sprintf("%s/sigma_rules", tenzirStorageFolder)
 	err = extractZIP("files.zip", sigmaPath)
 	if err != nil {
@@ -3311,7 +3298,6 @@ func handleFileCategoryChange() error {
 		return err
 	}
 
-
 	log.Printf("[DEBUG] Detection files copied to '%s' successfully.", sigmaPath)
 
 	return nil
@@ -3400,7 +3386,7 @@ func removeFileCategory() error {
 		tenzirStorageFolder = "/tmp/"
 	}
 
-	tenzirStorageFolder = strings.TrimRight(tenzirStorageFolder, "/") 
+	tenzirStorageFolder = strings.TrimRight(tenzirStorageFolder, "/")
 
 	//sigmaPath := "/var/lib/tenzir/sigma_rules/*"
 	sigmaPath := fmt.Sprintf("%s/sigma_rules", tenzirStorageFolder)
@@ -3565,9 +3551,8 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int {
 			return 0
 		}
 
-		labelSelector := "app=shuffle-worker"
 		pods, podErr := clientset.CoreV1().Pods(kubernetesNamespace).List(ctx, metav1.ListOptions{
-			LabelSelector: labelSelector,
+			LabelSelector: "app.kubernetes.io/name=shuffle-worker",
 		})
 		if podErr != nil {
 			log.Printf("[ERROR] Failed getting running workers: %s", podErr)

functions/onprem/worker/worker.go

@@ -23,8 +23,8 @@ import (
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
-	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/api/types/image"
+	"github.com/docker/docker/api/types/mount"
 	dockerclient "github.com/docker/docker/client"
 
 	// This is for automatic removal of certain code :)
@@ -482,9 +482,24 @@ func deployk8sApp(image string, identifier string, env []string) error {
 
 	//fix naming convention
 	// podUuid := uuid.NewV4().String()
-	// podName := fmt.Sprintf("%s-%s", value, podUuid)
+	// name := fmt.Sprintf("%s-%s", value, podUuid)
 	// replace identifier "_" with "-"
-	podName := strings.ReplaceAll(identifier, "_", "-")
+	name := strings.ReplaceAll(identifier, "_", "-")
+
+	labels := map[string]string{
+		"app.kubernetes.io/name":     "shuffle-app",
+		"app.kubernetes.io/instance": name,
+		// "app.kubernetes.io/version":    "",
+		"app.kubernetes.io/part-of":    "shuffle",
+		"app.kubernetes.io/managed-by": "shuffle-worker",
+		// Keep legacy labels for backward compatibility
+		"app": name,
+	}
+
+	matchLabels := map[string]string{
+		"app.kubernetes.io/name":     "shuffle-app",
+		"app.kubernetes.io/instance": name,
+	}
 
 	// pod := &corev1.Pod{
 	// 	ObjectMeta: metav1.ObjectMeta{
@@ -564,20 +579,17 @@ func deployk8sApp(image string, identifier string, env []string) error {
 
 	deployment := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: podName,
+			Name:   name,
+			Labels: labels,
 		},
 		Spec: appsv1.DeploymentSpec{
 			Replicas: &replicaNumberInt32,
 			Selector: &metav1.LabelSelector{
-				MatchLabels: map[string]string{
+				MatchLabels: matchLabels,
-					"app": podName,
-				},
 			},
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
-					Labels: map[string]string{
+					Labels: labels,
-						"app": podName,
-					},
 				},
 				Spec: corev1.PodSpec{
 					Containers: []corev1.Container{
@@ -601,12 +613,11 @@ func deployk8sApp(image string, identifier string, env []string) error {
 	// kubectl expose deployment {podName} --type=NodePort --port=80 --target-port=80
 	service := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: podName,
+			Name:   name,
+			Labels: labels,
 		},
 		Spec: corev1.ServiceSpec{
-			Selector: map[string]string{
+			Selector: matchLabels,
-				"app": podName,
-			},
 			Ports: []corev1.ServicePort{
 				{
 					Protocol:   "TCP",
@@ -954,7 +965,8 @@ 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)
-	labelSelector := fmt.Sprintf("app=shuffle-app,executionId=%s", workflowExecution.ExecutionId)
+	// FIXME: The executionId label is currently not set
+	labelSelector := fmt.Sprintf("app.kubernetes.io/name=shuffle-app,executionId=%s", workflowExecution.ExecutionId)
 
 	podList, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{
 		LabelSelector: labelSelector,