From 8d9409e184616fce797392e284d5c1dbd15c21f1 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Fri, 5 Jul 2024 05:10:13 +0530 Subject: [PATCH 01/11] feat[k8s]: adding a deployment layer --- functions/onprem/orborus/orborus.go | 174 ++++++++++++++++++++++------ 1 file changed, 141 insertions(+), 33 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index d1a09ff7..48617d08 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -51,9 +51,11 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + appsv1 "k8s.io/api/apps/v1" rbacv1 "k8s.io/api/rbac/v1" "k8s.io/apimachinery/pkg/util/intstr" - + // int32Ptr + ) // Starts jobs in bulk, so this could be increased @@ -218,7 +220,20 @@ func cleanupExistingNodes(ctx context.Context) error { } } - log.Printf("[INFO] Cleaned up all pods and services in namespace %s", kubernetesNamespace) + deployments, err := clientset.AppsV1().Deployments(kubernetesNamespace).List(context.Background(), metav1.ListOptions{}) + if err != nil { + log.Printf("[ERROR] Failed listing deployments: %s", err) + return err + } + + for _, deployment := range deployments.Items { + 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, services and deployments in namespace %s", kubernetesNamespace) return nil } @@ -832,51 +847,143 @@ func deployK8sWorker(image string, identifier string, env []string) error { // While testing: // kubectl delete pods --all --all-namespaces; kubectl delete services --all --all-namespaces - pod := &corev1.Pod{ + // pod := &corev1.Pod{ + // ObjectMeta: metav1.ObjectMeta{ + // Name: identifier, + // Labels: containerLabels, + // }, + // Spec: corev1.PodSpec{ + // RestartPolicy: "Never", + // // DNSPolicy: "Default", + // DNSPolicy: corev1.DNSClusterFirst, + // // NodeSelector: map[string]string{ + // // "node": "master", + // // }, + // Containers: []corev1.Container{ + // containerAttachment, + // }, + // }, + // } + + // // Check if running on ARM or x86 to download the correct image + + // // Get current pod's network so we can make the pod in it + + // _, err = clientset.CoreV1().Pods(kubernetesNamespace).List(context.Background(), metav1.ListOptions{}) + // if err != nil { + // log.Printf("[ERROR] Failed listing pods: %s", err) + // } + + + // createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{}) + // if err != nil { + // //log.Printf("[ERROR] Failed creating pod: %v", err) + // return err + // } + + // log.Printf("[INFO] Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace) + + // // kubectl expose pod shuffle-workers --type=LoadBalancer --port=33333 + // service := &corev1.Service{ + // ObjectMeta: metav1.ObjectMeta{ + // Name: identifier, + // }, + // Spec: corev1.ServiceSpec{ + // Selector: map[string]string{ + // "container": "shuffle-workers", + // }, + // Ports: []corev1.ServicePort{ + // { + // Protocol: "TCP", + // Port: 33333, + // TargetPort: intstr.FromInt(33333), + // }, + // }, + // Type: corev1.ServiceTypeLoadBalancer, + // }, + // } + + // _, 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 + + // experimenting with k8s deployments to enable autoscaling. + + // Spec: corev1.PodSpec{ + // RestartPolicy: "Never", + // // DNSPolicy: "Default", + // DNSPolicy: corev1.DNSClusterFirst, + // // NodeSelector: map[string]string{ + // // "node": "master", + // // }, + // Containers: []corev1.Container{ + // containerAttachment, + // }, + // }, + + deployment := &appsv1.Deployment{ ObjectMeta: metav1.ObjectMeta{ - Name: identifier, - Labels: containerLabels, + Name: identifier, }, - Spec: corev1.PodSpec{ - RestartPolicy: "Never", - // DNSPolicy: "Default", - DNSPolicy: corev1.DNSClusterFirst, - // NodeSelector: map[string]string{ - // "node": "master", - // }, - Containers: []corev1.Container{ - containerAttachment, + Spec: appsv1.DeploymentSpec{ + Replicas: int32Ptr(1), + Selector: &metav1.LabelSelector{ + MatchLabels: containerLabels, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: containerLabels, + }, + Spec: corev1.PodSpec{ + DNSPolicy: corev1.DNSClusterFirst, + // NodeSelector: map[string]string{ + // "node": "master", + // }, + Containers: []corev1.Container{ + containerAttachment, + }, + }, }, }, } - // Check if running on ARM or x86 to download the correct image - - // Get current pod's network so we can make the pod in it - - _, err = clientset.CoreV1().Pods(kubernetesNamespace).List(context.Background(), metav1.ListOptions{}) + _, err = clientset.AppsV1().Deployments(kubernetesNamespace).Create(context.Background(), deployment, metav1.CreateOptions{}) if err != nil { - log.Printf("[ERROR] Failed listing pods: %s", err) - } - - - 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 deployment: %v", err) return err } - log.Printf("[INFO] Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace) + // // kubectl expose pod shuffle-workers --type=LoadBalancer --port=33333 + // service := &corev1.Service{ + // ObjectMeta: metav1.ObjectMeta{ + // Name: identifier, + // }, + // Spec: corev1.ServiceSpec{ + // Selector: map[string]string{ + // "container": "shuffle-workers", + // }, + // Ports: []corev1.ServicePort{ + // { + // Protocol: "TCP", + // Port: 33333, + // TargetPort: intstr.FromInt(33333), + // }, + // }, + // Type: corev1.ServiceTypeLoadBalancer, + // }, + // } - // kubectl expose pod shuffle-workers --type=LoadBalancer --port=33333 + // kubectl expose deployment shuffle-workers --type=NodePort --port=33333 service := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: identifier, }, Spec: corev1.ServiceSpec{ - Selector: map[string]string{ - "container": "shuffle-workers", - }, + Selector: containerLabels, Ports: []corev1.ServicePort{ { Protocol: "TCP", @@ -884,11 +991,11 @@ func deployK8sWorker(image string, identifier string, env []string) error { TargetPort: intstr.FromInt(33333), }, }, - Type: corev1.ServiceTypeLoadBalancer, + Type: corev1.ServiceTypeNodePort, }, } - _, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.TODO(), service, metav1.CreateOptions{}) + _, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.Background(), service, metav1.CreateOptions{}) if err != nil { log.Printf("[ERROR] Failed creating service: %v", err) return err @@ -897,6 +1004,7 @@ func deployK8sWorker(image string, identifier string, env []string) error { return nil } +func int32Ptr(i int32) *int32 { return &i } func deployWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error { if len(os.Getenv("REGISTRY_URL")) > 0 && os.Getenv("REGISTRY_URL") != "" { From 178c03e255196f754fc30b937c7a13f8b15c01b6 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Fri, 5 Jul 2024 17:52:30 +0530 Subject: [PATCH 02/11] fix: push before i try to do replica based scaling --- functions/onprem/orborus/orborus.go | 601 +++++++++++++--------------- 1 file changed, 274 insertions(+), 327 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 48617d08..02a57740 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -49,13 +49,11 @@ import ( //"github.com/mackerelio/go-osstat/memory" //"github.com/shirou/gopsutil/cpu" - corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" - // int32Ptr - ) // Starts jobs in bulk, so this could be increased @@ -77,7 +75,6 @@ var isKubernetes = os.Getenv("IS_KUBERNETES") var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") var maxCPUPercent = 90 - // var baseimagename = "docker.pkg.github.com/shuffle/shuffle" // var baseimagename = "ghcr.io/frikky" // var baseimagename = "shuffle/shuffle" @@ -106,7 +103,7 @@ var memcached = os.Getenv("SHUFFLE_MEMCACHED") var tenzirUrl = os.Getenv("SHUFFLE_TENZIR_URL") var executionIds = []string{} -var namespacemade = false // For K8s +var namespacemade = false // For K8s var dockercli *dockerclient.Client var containerId string @@ -233,11 +230,13 @@ func cleanupExistingNodes(ctx context.Context) error { } } - log.Printf("[INFO] Cleaned up all pods, services and deployments in namespace %s", kubernetesNamespace) + log.Printf("[INFO] Cleaned up all pods and services in namespace %s. Waiting 10 seconds for cleanup to reflect", kubernetesNamespace) + + time.Sleep(10 * time.Second) + return nil } - serviceListOptions := types.ServiceListOptions{} services, err := dockercli.ServiceList( context.Background(), @@ -687,11 +686,10 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar { return envVars } - func handleBackendImageDownload(ctx context.Context, images string) error { // Should use docker to: - // 1. Pull the image & tag it - // 2. Distribute the image by updating service if "run" + // 1. Pull the image & tag it + // 2. Distribute the image by updating service if "run" if swarmConfig == "run" || swarmConfig == "swarm" { log.Printf("[DEBUG] Should update service with new image after updating(s): %s. \n\nNOT IMPLEMENTED: Contact support@shuffler.io for support.\n\n", images) @@ -702,8 +700,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error { log.Printf("[DEBUG] Should remove existing image (s): %s", images) // Remove the image - removeOptions := image.RemoveOptions{ - } + removeOptions := image.RemoveOptions{} for _, image := range strings.Split(images, ",") { image = strings.TrimSpace(image) @@ -723,6 +720,117 @@ func handleBackendImageDownload(ctx context.Context, images string) error { return nil } +func fixk8sRoles() { + clientset, _, err := shuffle.GetKubernetesClient() + if err != nil { + log.Printf("[ERROR] Error getting kubernetes client: %s", err) + os.Exit(1) + } + + kubernetesNamespace := "default" + + // Check if namespace exist as variable. If so, make it + if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 { + kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") + } + + // fix roles + // check if "service-creator" role is assigned to the service account "default" + // roleBindingNames := []string{"service-creator-binding", "pod-creator-binding", "deployment-creator-binding"} + serviceAccountName := "default" + roleBindingName := "creator-all" + + resourceTypes := []string{"services", "pods", "deployments"} + + // Check if the RoleBinding exists + roleBinding, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Get(context.TODO(), roleBindingName, metav1.GetOptions{}) + if err != nil { + log.Printf("[WARNING] Failed to get RoleBinding %s: %s", roleBindingName, err) + + // create role and rolebinding + role := &rbacv1.Role{ + ObjectMeta: metav1.ObjectMeta{ + Name: roleBindingName, + }, + Rules: []rbacv1.PolicyRule{ + { + APIGroups: []string{""}, + Resources: resourceTypes, + Verbs: []string{"create", "list"}, + }, + }, + } + + ctx := context.TODO() + + _, err := clientset.RbacV1().Roles(kubernetesNamespace).Create(ctx, role, metav1.CreateOptions{}) + if err != nil { + log.Printf("[ERROR] Failed to create Role %s: %s", roleBindingName, err) + if !strings.Contains(fmt.Sprintf("%s", err), "already exists") { + log.Printf("[INFO] role %s already exists", roleBindingName) + } + } + + roleBinding := &rbacv1.RoleBinding{ + ObjectMeta: metav1.ObjectMeta{ + Name: roleBindingName, + }, + Subjects: []rbacv1.Subject{ + { + Kind: "ServiceAccount", + Name: serviceAccountName, + Namespace: kubernetesNamespace, + }, + }, + RoleRef: rbacv1.RoleRef{ + Kind: "Role", + Name: roleBindingName, + }, + } + + _, err = clientset.RbacV1().RoleBindings(kubernetesNamespace).Create(ctx, roleBinding, metav1.CreateOptions{}) + if err != nil { + log.Printf("[ERROR] Failed to create RoleBinding %s: %s", roleBindingName, err) + if strings.Contains(fmt.Sprintf("%s", err), "already exists") { + log.Printf("[INFO] rolebinding %s already exists", roleBindingName) + } + } + + log.Printf("[INFO] Created Role %s and RoleBinding %s", roleBindingName, roleBindingName) + } else { + log.Printf("[INFO] RoleBinding %s exists", roleBindingName) + } + + // Check if the RoleBinding is assigned to the service account + var found bool + for _, subject := range roleBinding.Subjects { + if subject.Kind == "ServiceAccount" && subject.Name == serviceAccountName { + found = true + break + } + } + + if !found { + log.Printf("[WARNING] Service account %s is not assigned to RoleBinding %s\n", serviceAccountName, roleBindingName) + // assign the service account to the rolebinding + roleBinding.Subjects = append(roleBinding.Subjects, rbacv1.Subject{ + Kind: "ServiceAccount", + Name: serviceAccountName, + Namespace: kubernetesNamespace, + }) + + ctx := context.TODO() + + _, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Update(ctx, roleBinding, metav1.UpdateOptions{}) + if err != nil { + log.Printf("[ERROR](ns - %s) Failed to update RoleBinding %s: %s", kubernetesNamespace, roleBindingName, err) + if !strings.Contains(fmt.Sprintf("%s", err), "already exists") { + log.Printf("[INFO] rolebinding %s already exists", roleBindingName) + } + } + } +} + func deployK8sWorker(image string, identifier string, env []string) error { env = append(env, fmt.Sprintf("IS_KUBERNETES=true")) env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE"))) @@ -748,7 +856,7 @@ func deployK8sWorker(image string, identifier string, env []string) error { //env = append(env, fmt.Sprintf("KUBERNETES_CONFIG=%s", config.String())) // FIXME: When a service account is used, the account is also mounted in the pod - // The volume mount location is: + // The volume mount location is: // /var/run/secrets/kubernetes.io/serviceaccount // Look for if there is a default service account in use @@ -759,7 +867,6 @@ func deployK8sWorker(image string, identifier string, env []string) error { // use k8s downward API to find it if we are in a pod } - // Check if namespace exist as variable. If so, make it if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 && !namespacemade { kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") @@ -785,9 +892,10 @@ func deployK8sWorker(image string, identifier string, env []string) error { env = append(env, fmt.Sprintf("BASE_URL=%s", baseUrl)) env = append(env, fmt.Sprintf("SHUFFLE_SWARM_CONFIG=%s", swarmConfig)) + env = append(env, fmt.Sprintf("WORKER_HOSTNAME=%s", "shuffle-workers")) if len(kubernetesNamespace) == 0 { - foundNamespace, err := shuffle.GetKubernetesNamespace() + foundNamespace, err := shuffle.GetKubernetesNamespace() if err != nil { //log.Printf("[ERROR] Failed getting Kubernetes namespace: %s", err) } @@ -825,7 +933,7 @@ func deployK8sWorker(image string, identifier string, env []string) error { Name: identifier, Image: kubernetesImage, Env: buildEnvVars(envMap), - + //ImagePullPolicy: "Never", ImagePullPolicy: corev1.PullIfNotPresent, } @@ -844,7 +952,6 @@ func deployK8sWorker(image string, identifier string, env []string) error { } } - // While testing: // kubectl delete pods --all --all-namespaces; kubectl delete services --all --all-namespaces // pod := &corev1.Pod{ @@ -874,7 +981,6 @@ func deployK8sWorker(image string, identifier string, env []string) error { // log.Printf("[ERROR] Failed listing pods: %s", err) // } - // createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{}) // if err != nil { // //log.Printf("[ERROR] Failed creating pod: %v", err) @@ -909,28 +1015,11 @@ func deployK8sWorker(image string, identifier string, env []string) error { // return err // } - // return nil - - // experimenting with k8s deployments to enable autoscaling. - - // Spec: corev1.PodSpec{ - // RestartPolicy: "Never", - // // DNSPolicy: "Default", - // DNSPolicy: corev1.DNSClusterFirst, - // // NodeSelector: map[string]string{ - // // "node": "master", - // // }, - // Containers: []corev1.Container{ - // containerAttachment, - // }, - // }, - deployment := &appsv1.Deployment{ ObjectMeta: metav1.ObjectMeta{ Name: identifier, }, Spec: appsv1.DeploymentSpec{ - Replicas: int32Ptr(1), Selector: &metav1.LabelSelector{ MatchLabels: containerLabels, }, @@ -939,13 +1028,10 @@ func deployK8sWorker(image string, identifier string, env []string) error { Labels: containerLabels, }, Spec: corev1.PodSpec{ - DNSPolicy: corev1.DNSClusterFirst, - // NodeSelector: map[string]string{ - // "node": "master", - // }, Containers: []corev1.Container{ containerAttachment, }, + DNSPolicy: corev1.DNSClusterFirst, }, }, }, @@ -957,27 +1043,7 @@ func deployK8sWorker(image string, identifier string, env []string) error { return err } - // // kubectl expose pod shuffle-workers --type=LoadBalancer --port=33333 - // service := &corev1.Service{ - // ObjectMeta: metav1.ObjectMeta{ - // Name: identifier, - // }, - // Spec: corev1.ServiceSpec{ - // Selector: map[string]string{ - // "container": "shuffle-workers", - // }, - // Ports: []corev1.ServicePort{ - // { - // Protocol: "TCP", - // Port: 33333, - // TargetPort: intstr.FromInt(33333), - // }, - // }, - // Type: corev1.ServiceTypeLoadBalancer, - // }, - // } - - // kubectl expose deployment shuffle-workers --type=NodePort --port=33333 + // kubectl expose deployment shuffle-workers --type=NodePort --port=33333 --target-port=33333 service := &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: identifier, @@ -986,8 +1052,8 @@ func deployK8sWorker(image string, identifier string, env []string) error { Selector: containerLabels, Ports: []corev1.ServicePort{ { - Protocol: "TCP", - Port: 33333, + Protocol: "TCP", + Port: 33333, TargetPort: intstr.FromInt(33333), }, }, @@ -1004,8 +1070,6 @@ func deployK8sWorker(image string, identifier string, env []string) error { return nil } -func int32Ptr(i int32) *int32 { return &i } - 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"))) @@ -1407,8 +1471,7 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats { newStats.MaxMemory = int(pers.MemTotal) } - - // Get list of all running containers + // Get list of all running containers containers, err := dockercli.ContainerList(ctx, container.ListOptions{}) if err != nil { @@ -1522,10 +1585,6 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats { return newStats } - - - - func sendRemoveRequest(client *http.Client, toBeRemoved shuffle.ExecutionRequestWrapper, baseUrl, environment, auth, org string, sleepTime int) error { confirmUrl := fmt.Sprintf("%s/api/v1/workflows/queue/confirm", baseUrl) @@ -1604,112 +1663,7 @@ func main() { } if isKubernetes == "true" { - clientset, _, err := shuffle.GetKubernetesClient() - if err != nil { - log.Printf("[ERROR] Error getting kubernetes client: %s", err) - os.Exit(1) - } - - kubernetesNamespace := "default" - - // Check if namespace exist as variable. If so, make it - if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 && !namespacemade { - kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") - } - - // fix roles - // check if "service-creator" role is assigned to the service account "default" - roleBindingName := "service-creator-binding" - serviceAccountName := "default" - // Check if the RoleBinding exists - roleBinding, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Get(context.TODO(), roleBindingName, metav1.GetOptions{}) - if err != nil { - log.Printf("[WARNING] Failed to get RoleBinding %s: %s", roleBindingName, err) - // create role and rolebinding - role := &rbacv1.Role{ - ObjectMeta: metav1.ObjectMeta{ - Name: roleBindingName, - }, - Rules: []rbacv1.PolicyRule{ - { - APIGroups: []string{""}, - Resources: []string{"services"}, - Verbs: []string{"create"}, - }, - }, - } - - ctx := context.TODO() - - _, err := clientset.RbacV1().Roles(kubernetesNamespace).Create(ctx, role, metav1.CreateOptions{}) - if err != nil { - log.Printf("[ERROR] Failed to create Role %s: %s", roleBindingName, err) - if !strings.Contains(fmt.Sprintf("%s", err), "already exists") { - log.Printf("[INFO] role %s already exists", roleBindingName) - } - } - - roleBinding := &rbacv1.RoleBinding{ - ObjectMeta: metav1.ObjectMeta{ - Name: roleBindingName, - }, - Subjects: []rbacv1.Subject{ - { - Kind: "ServiceAccount", - Name: serviceAccountName, - Namespace: kubernetesNamespace, - }, - }, - RoleRef: rbacv1.RoleRef{ - Kind: "Role", - Name: roleBindingName, - }, - } - - - - _, err = clientset.RbacV1().RoleBindings(kubernetesNamespace).Create(ctx, roleBinding, metav1.CreateOptions{}) - if err != nil { - log.Printf("[ERROR] Failed to create RoleBinding %s: %s", roleBindingName, err) - if !strings.Contains(fmt.Sprintf("%s", err), "already exists") { - log.Printf("[INFO] rolebinding %s already exists", roleBindingName) - } - } - - - log.Printf("[INFO] Created Role %s and RoleBinding %s", roleBindingName, roleBindingName) - } else { - log.Printf("[INFO] RoleBinding %s exists", roleBindingName) - } - - // Check if the RoleBinding is assigned to the service account - var found bool - for _, subject := range roleBinding.Subjects { - if subject.Kind == "ServiceAccount" && subject.Name == serviceAccountName { - found = true - break - } - } - - if !found { - log.Printf("[WARNING] Service account %s is not assigned to RoleBinding %s\n", serviceAccountName, roleBindingName) - // assign the service account to the rolebinding - roleBinding.Subjects = append(roleBinding.Subjects, rbacv1.Subject{ - Kind: "ServiceAccount", - Name: serviceAccountName, - Namespace: kubernetesNamespace, - }) - - ctx := context.TODO() - - _, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Update(ctx, roleBinding, metav1.UpdateOptions{}) - if err != nil { - log.Printf("[ERROR] Failed to update RoleBinding %s: %s", roleBindingName, err) - if !strings.Contains(fmt.Sprintf("%s", err), "already exists") { - log.Printf("[INFO] rolebinding %s already exists", roleBindingName) - } - } - } + fixk8sRoles() } startupDelay := os.Getenv("SHUFFLE_ORBORUS_STARTUP_DELAY") @@ -1819,7 +1773,7 @@ func main() { log.Printf("[DEBUG] Cleaning up containers from previous run") cleanupExistingNodes(ctx) - time.Sleep(time.Duration(5) * time.Second) + time.Sleep(time.Duration(5) * time.Second) log.Printf("[DEBUG] Deploying worker image %s to swarm", workerImage) @@ -1987,13 +1941,12 @@ func main() { continue } - if hasStarted && len(executionRequests.Data) > 0 { //log.Printf("[INFO] Body: %s", string(body)) // Type string `json:"type"` } - // FIXME: Add features here for orborus & worker to + // FIXME: Add features here for orborus & worker to // do things on behalf of backend var toBeRemoved shuffle.ExecutionRequestWrapper if len(executionRequests.Data) > 0 { @@ -2065,7 +2018,7 @@ func main() { log.Printf("[WARNING] Throttle - Cutting down requests from %d to %d (MAX: %d, CUR: %d)", len(executionRequests.Data), allowed, maxConcurrency, executionCount) executionRequests.Data = executionRequests.Data[0:allowed] } - } else if (swarmControlMode && (swarmConfig == "run" || swarmConfig == "swarm")) { + } else if swarmControlMode && (swarmConfig == "run" || swarmConfig == "swarm") { if len(executionRequests.Data) > 50 { executionRequests.Data = executionRequests.Data[0:50] } @@ -2214,7 +2167,6 @@ func main() { } } - // func deployPipeline(image, identifier, command string) error { // if isKubernetes == "true" { // return errors.New("Kubernetes not implemented") @@ -2239,7 +2191,6 @@ func main() { // envVariables := []string{ // } - // // Add volume binds for storage // // Want read/write with full access for the container // //sourceFolder := "/Users/frikky/git/shuffle/shuffle-database" @@ -2276,8 +2227,7 @@ func main() { // config.Labels = map[string]string{ // "name": identifier, // "shuffle": "shuffle", -// } - +// } // cont, err := dockercli.ContainerCreate( // ctx, @@ -2299,8 +2249,8 @@ func main() { // containerStartOptions := container.StartOptions{} // err = dockercli.ContainerStart( -// ctx, -// cont.ID, +// ctx, +// cont.ID, // containerStartOptions, // ) // if err != nil { @@ -2319,8 +2269,8 @@ func main() { // } // err = dockercli.ContainerStart( -// ctx, -// cont.ID, +// ctx, +// cont.ID, // containerStartOptions, // ) // if err != nil { @@ -2364,8 +2314,6 @@ func main() { // return nil // } - - // Tenzir command samples // docker pull ghcr.io/dominiklohmann/tenzir-arm64:latest // docker tag ghcr.io/dominiklohmann/tenzir-arm64:latest tenzir/tenzir:latest @@ -2373,14 +2321,14 @@ func main() { // Read from Cache and send it to a webhook // docker run tenzir/tenzir:latest 'from http://192.168.86.44:5002/api/v1/orgs/7e9b9007-5df2-4b47-bca5-c4d267ef2943/cache/CIDR%20ranges?type=text&authorization=cec9d01f-09b2-4419-8a0a-76c6046e3fef read lines | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines' func handlePipeline(incRequest shuffle.ExecutionRequest) error { - + if tenzirUrl == "" { tenzirUrl = "http://localhost:5160" - log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s",tenzirUrl) + log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s", tenzirUrl) } err := deployTenzirNode() - if err != nil{ + if err != nil { log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err) return err } @@ -2414,7 +2362,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { if err != nil { log.Printf("[ERROR] Failed Deleting Pipeline %s", err) return err - } + } } else if incRequest.Type == "PIPELINE_STOP" { log.Printf("[INFO] Should stop the pipeline %#v", identifier) pipelineId, err := searchPipeline(identifier) @@ -2430,10 +2378,10 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { log.Printf("[INFO] successfully stopped the Pipeline: %s", pipelineId) } - } else if incRequest.Type == "PIPELINE_START" { + } else if incRequest.Type == "PIPELINE_START" { log.Printf("[INFO] Should start the pipeline %#v", identifier) pipelineId, err := searchPipeline(identifier) - if err != nil { + if err != nil { if err.Error() == "no existing pipeline found with name" { log.Printf("[WARNING] no pipeline found for %s, creating a new one", identifier) _, CreateErr := createPipeline(command, identifier) @@ -2459,157 +2407,157 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { } func deployTenzirNode() error { - if isKubernetes == "true" { - return errors.New("kubernetes not implemented") - } + if isKubernetes == "true" { + return errors.New("kubernetes not implemented") + } - ctx := context.Background() - cacheKey := "tenzir-key" + ctx := context.Background() + cacheKey := "tenzir-key" - imageName := "tenzir/tenzir:latest" - containerName := "tenzir-node" - containerStartOptions := container.StartOptions{} + imageName := "tenzir/tenzir:latest" + containerName := "tenzir-node" + containerStartOptions := container.StartOptions{} - _, err := shuffle.GetCache(ctx, cacheKey) - if err == nil { - return nil - } + _, err := shuffle.GetCache(ctx, cacheKey) + if err == nil { + return nil + } - containerInfo, err := dockercli.ContainerInspect(ctx, containerName) - if err != nil { - if dockerclient.IsErrNotFound(err) { + containerInfo, err := dockercli.ContainerInspect(ctx, containerName) + if err != nil { + if dockerclient.IsErrNotFound(err) { - // Check if image exists - _, _, err := dockercli.ImageInspectWithRaw(ctx, imageName) - if dockerclient.IsErrNotFound(err) { - log.Printf("[DEBUG] pulling image %s", imageName) - pullOptions := image.PullOptions{} - out, err := dockercli.ImagePull(ctx, imageName, pullOptions) - if err != nil { - log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err) - return err - } - defer out.Close() + // Check if image exists + _, _, err := dockercli.ImageInspectWithRaw(ctx, imageName) + if dockerclient.IsErrNotFound(err) { + log.Printf("[DEBUG] pulling image %s", imageName) + pullOptions := image.PullOptions{} + out, err := dockercli.ImagePull(ctx, imageName, pullOptions) + if err != nil { + log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err) + return err + } + defer out.Close() - io.Copy(io.Discard, out) - } else if err != nil { - return err - } + io.Copy(io.Discard, out) + } else if err != nil { + return err + } - err = createAndStartTenzirNode(ctx, containerName, imageName, containerStartOptions) - if err != nil { - return err - } - } else { - return err - } - } else { - if !containerInfo.State.Running { - log.Printf("[DEBUG] Tenzir Node exists but is not running") - err := dockercli.ContainerStart(ctx, containerName, containerStartOptions) - if err != nil { - log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err) - return err - } + err = createAndStartTenzirNode(ctx, containerName, imageName, containerStartOptions) + if err != nil { + return err + } + } else { + return err + } + } else { + if !containerInfo.State.Running { + log.Printf("[DEBUG] Tenzir Node exists but is not running") + err := dockercli.ContainerStart(ctx, containerName, containerStartOptions) + if err != nil { + log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err) + return err + } - log.Printf("[INFO] Waiting for Tenzir to become available ...") - err = checkTenzirNode() - if err != nil { - return err - } - } - } + log.Printf("[INFO] Waiting for Tenzir to become available ...") + err = checkTenzirNode() + if err != nil { + return err + } + } + } - tenzirStatus := struct { - ContainerStatus string `json:"container_status"` - }{ - ContainerStatus: "running", - } + tenzirStatus := struct { + ContainerStatus string `json:"container_status"` + }{ + ContainerStatus: "running", + } - cacheData, err := json.Marshal(tenzirStatus) - if err != nil { - log.Printf("[WARNING] Failed marshalling execution: %s", err) - } - err = shuffle.SetCache(ctx, cacheKey, cacheData, 1) - if err != nil { - log.Printf("[WARNING] Failed updating cache for tenzir: %s", err) - } + cacheData, err := json.Marshal(tenzirStatus) + if err != nil { + log.Printf("[WARNING] Failed marshalling execution: %s", err) + } + err = shuffle.SetCache(ctx, cacheKey, cacheData, 1) + if err != nil { + log.Printf("[WARNING] Failed updating cache for tenzir: %s", err) + } - return nil + return nil } func checkTenzirNode() error { - retries := 20 - retryInterval := 3 * time.Second - url := fmt.Sprintf("%s/api/v0/ping",tenzirUrl) + retries := 20 + retryInterval := 3 * time.Second + url := fmt.Sprintf("%s/api/v0/ping", tenzirUrl) forwardMethod := "POST" - client := http.Client{} - req, err := http.NewRequest(forwardMethod, url, nil) + client := http.Client{} + req, err := http.NewRequest(forwardMethod, url, nil) if err != nil { log.Printf("[ERROR] Failed to create HTTP request: %s", err) return err } - for i := 0; i < retries; i++ { - resp, err := client.Do(req) - if err == nil && resp.StatusCode == http.StatusOK { - return nil - } - time.Sleep(retryInterval) - } + for i := 0; i < retries; i++ { + resp, err := client.Do(req) + if err == nil && resp.StatusCode == http.StatusOK { + return nil + } + time.Sleep(retryInterval) + } - return fmt.Errorf("tenzir node is not available") + return fmt.Errorf("tenzir node is not available") } func createAndStartTenzirNode(ctx context.Context, containerName, imageName string, containerStartOptions container.StartOptions) error { - healthconfig := &container.HealthConfig{ - Test: []string{"tenzir --connection-timeout=30s --connection-retry-delay=1s 'api /ping'"}, - Interval: 30 * time.Second, - Retries: 1, - } + healthconfig := &container.HealthConfig{ + Test: []string{"tenzir --connection-timeout=30s --connection-retry-delay=1s 'api /ping'"}, + Interval: 30 * time.Second, + Retries: 1, + } - config := &container.Config{ - Cmd: []string{"--commands=web server --mode=dev --bind=0.0.0.0"}, - Image: imageName, - Healthcheck: healthconfig, - ExposedPorts: nat.PortSet{"5160/tcp": struct{}{}}, - Entrypoint: []string{containerName}, - } + config := &container.Config{ + Cmd: []string{"--commands=web server --mode=dev --bind=0.0.0.0"}, + Image: imageName, + Healthcheck: healthconfig, + ExposedPorts: nat.PortSet{"5160/tcp": struct{}{}}, + Entrypoint: []string{containerName}, + } - hostConfig := &container.HostConfig{ - PortBindings: nat.PortMap{ - "5160/tcp": []nat.PortBinding{{HostPort: "5160"}}, - }, - Mounts: []mount.Mount{ - { - Type: mount.TypeVolume, - Source: containerName, - Target: "/var/lib/tenzir/", - }, - }, - VolumeDriver: "local", - } - _, err := dockercli.ContainerCreate(ctx, config, hostConfig, nil, nil, containerName) - if err != nil { - return err - } + hostConfig := &container.HostConfig{ + PortBindings: nat.PortMap{ + "5160/tcp": []nat.PortBinding{{HostPort: "5160"}}, + }, + Mounts: []mount.Mount{ + { + Type: mount.TypeVolume, + Source: containerName, + Target: "/var/lib/tenzir/", + }, + }, + VolumeDriver: "local", + } + _, err := dockercli.ContainerCreate(ctx, config, hostConfig, nil, nil, containerName) + if err != nil { + return err + } - err = dockercli.ContainerStart(ctx, containerName, containerStartOptions) - if err != nil { - log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err) - return err - } - log.Printf("[INFO] Tenzir Node container started successfully") + err = dockercli.ContainerStart(ctx, containerName, containerStartOptions) + if err != nil { + log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err) + return err + } + log.Printf("[INFO] Tenzir Node container started successfully") - log.Printf("[INFO] Waiting for Tenzir to become available ...") - err = checkTenzirNode() - if err != nil { - return err - } - log.Printf("[INFO] Successfully deployed Tenzir Node !") + log.Printf("[INFO] Waiting for Tenzir to become available ...") + err = checkTenzirNode() + if err != nil { + return err + } + log.Printf("[INFO] Successfully deployed Tenzir Node !") - return nil + return nil } func createPipeline(command, identifier string) (string, error) { @@ -2617,7 +2565,7 @@ func createPipeline(command, identifier string) (string, error) { toBeDeleted := false pipelineId, err := searchPipeline(identifier) - url := fmt.Sprintf("%s/api/v0/pipeline/create", tenzirUrl) + url := fmt.Sprintf("%s/api/v0/pipeline/create", tenzirUrl) forwardMethod := "POST" if err != nil { @@ -2630,7 +2578,7 @@ func createPipeline(command, identifier string) (string, error) { log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId) toBeDeleted = true } - if strings.Contains(command, "shuffler.io") { + if strings.Contains(command, "shuffler.io") { } else { var scheme string @@ -2644,7 +2592,7 @@ func createPipeline(command, identifier string) (string, error) { if startIndex != -1 { endIndex := startIndex + len(scheme) endIndex += strings.Index(command[endIndex:], "/") - + command = command[:startIndex] + baseUrl + command[endIndex:] } } @@ -2723,7 +2671,7 @@ func createPipeline(command, identifier string) (string, error) { func updatePipelineState(pipelineId, action string) (string, error) { - url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl) + url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl) forwardMethod := "POST" requestBody := map[string]interface{}{ @@ -2792,7 +2740,7 @@ func deletePipeline(pipelineId string) error { "id": pipelineId, } - url := fmt.Sprintf("%s/api/v0/pipeline/delete", tenzirUrl) + url := fmt.Sprintf("%s/api/v0/pipeline/delete", tenzirUrl) forwardMethod := "POST" requestBodyJSON, err := json.Marshal(requestBody) @@ -2838,9 +2786,9 @@ func searchPipeline(identifier string) (string, error) { Name string `json:"name"` } - var reqBody []byte + var reqBody []byte - url := fmt.Sprintf("%s/api/v0/pipeline/list", tenzirUrl) + url := fmt.Sprintf("%s/api/v0/pipeline/list", tenzirUrl) resp, err := http.Post(url, "application/json", bytes.NewBuffer(reqBody)) if err != nil { @@ -2925,10 +2873,9 @@ func searchPipeline(identifier string) (string, error) { func getRunningWorkers(ctx context.Context, workerTimeout int) int { //log.Printf("[DEBUG] Getting running workers with API version %s", dockerApiVersion) counter := 0 - if isKubernetes == "true" { + if isKubernetes == "true" { log.Printf("[INFO] Getting running workers in kubernetes") - thresholdTime := time.Now().Add(time.Duration(-workerTimeout) * time.Second) clientset, _, err := shuffle.GetKubernetesClient() From 5448c1fe6d671834240c79703e20e0d4cd62fd31 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Fri, 5 Jul 2024 18:12:51 +0530 Subject: [PATCH 03/11] fix[k8s]: worker replication works at scale --- functions/onprem/orborus/orborus.go | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 02a57740..ffc306e6 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -831,6 +831,9 @@ func fixk8sRoles() { } } + +func int32Ptr(i int32) *int32 { return &i } + func deployK8sWorker(image string, identifier string, env []string) error { env = append(env, fmt.Sprintf("IS_KUBERNETES=true")) env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE"))) @@ -1015,11 +1018,25 @@ func deployK8sWorker(image string, identifier string, env []string) error { // return err // } + 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: identifier, }, Spec: appsv1.DeploymentSpec{ + Replicas: int32Ptr(replicaNumberInt32), Selector: &metav1.LabelSelector{ MatchLabels: containerLabels, }, From fb7b32682d439cf6f0a0d8931f4bcc2c4b9b969c Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Fri, 5 Jul 2024 19:14:30 +0530 Subject: [PATCH 04/11] fix[k8s]: app replication works at scale --- functions/onprem/orborus/orborus.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index ffc306e6..b1688487 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -754,7 +754,7 @@ func fixk8sRoles() { }, Rules: []rbacv1.PolicyRule{ { - APIGroups: []string{""}, + APIGroups: []string{"", "apps"}, Resources: resourceTypes, Verbs: []string{"create", "list"}, }, @@ -1026,6 +1026,7 @@ func deployK8sWorker(image string, identifier string, env []string) error { log.Printf("[ERROR] %s is not a valid number for replication", replicaNumberStr) } else { replicaNumber = tmpInt + } } @@ -3200,7 +3201,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string, debugCommand := fmt.Sprintf("docker service logs shuffle-workers 2>&1 -f | grep %s", workflowExecution.ExecutionId) if isKubernetes == "true" { - debugCommand = fmt.Sprintf("kubectl logs -n %s %s | grep %s", kubernetesNamespace, identifier, workflowExecution.ExecutionId) + debugCommand = fmt.Sprintf("kubectl logs -n %s container=shuffle-worker | grep %s", kubernetesNamespace, workflowExecution.ExecutionId) } log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s. DEBUGGING:\n%s", workflowExecution.ExecutionId, streamUrl, debugCommand) From 025489a5c7afd5ad06152fd271ebfce8fd8fb1fe Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Tue, 9 Jul 2024 17:40:31 +0530 Subject: [PATCH 05/11] fix[k8s]: keeping old core pods (orborus, frontend, backend etc) --- functions/onprem/orborus/orborus.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index b1688487..3fbf16b7 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -179,6 +179,9 @@ func getThisContainerId() { func cleanupExistingNodes(ctx context.Context) error { if isKubernetes == "true" { + // of course, this doesn't clean up "nodes" but + // rather pods, services, roles etc. + if kubernetesNamespace == "" { kubernetesNamespace = "default" } @@ -197,6 +200,16 @@ func cleanupExistingNodes(ctx context.Context) error { } for _, pod := range pods.Items { + // check if pod.Name starts with: + // "backend-", "frontend-", "orborus-", "opensearch-" or "memcached-" + if strings.HasPrefix(pod.Name, "backend-") || + strings.HasPrefix(pod.Name, "frontend-") || + strings.HasPrefix(pod.Name, "orborus-") || + strings.HasPrefix(pod.Name, "opensearch-") || + strings.HasPrefix(pod.Name, "memcached-") { + 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) @@ -211,6 +224,13 @@ func cleanupExistingNodes(ctx context.Context) error { } for _, service := range services.Items { + if strings.Contains(service.Name, "opensearch") || + strings.Contains(service.Name, "memcached") || + strings.Contains(service.Name, "shuffle-backend") || strings.Contains(service.Name, "backend") || + strings.Contains(service.Name, "frontend") { + 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) From 22f9987f4111a0f6e603464a63ba5d3291e02fc9 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Tue, 9 Jul 2024 17:48:43 +0530 Subject: [PATCH 06/11] fix[k8s]: keeping old core pods (orborus, frontend, backend etc) [with better code] --- functions/onprem/orborus/orborus.go | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 3fbf16b7..4cfc603e 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -176,6 +176,17 @@ func getThisContainerId() { log.Printf(`[INFO] Started with containerId "%s"`, containerId) } + +func skipCheckInCleanup(name string) bool { + + return strings.HasPrefix(name, "backend-") || + strings.HasPrefix(name, "shuffle-backend") || + strings.HasPrefix(name, "frontend-") || + strings.HasPrefix(name, "orborus-") || + strings.HasPrefix(name, "opensearch-") || + strings.HasPrefix(name, "memcached-") +} + func cleanupExistingNodes(ctx context.Context) error { if isKubernetes == "true" { @@ -202,11 +213,7 @@ func cleanupExistingNodes(ctx context.Context) error { for _, pod := range pods.Items { // check if pod.Name starts with: // "backend-", "frontend-", "orborus-", "opensearch-" or "memcached-" - if strings.HasPrefix(pod.Name, "backend-") || - strings.HasPrefix(pod.Name, "frontend-") || - strings.HasPrefix(pod.Name, "orborus-") || - strings.HasPrefix(pod.Name, "opensearch-") || - strings.HasPrefix(pod.Name, "memcached-") { + if skipCheckInCleanup(pod.Name) { continue } @@ -224,10 +231,7 @@ func cleanupExistingNodes(ctx context.Context) error { } for _, service := range services.Items { - if strings.Contains(service.Name, "opensearch") || - strings.Contains(service.Name, "memcached") || - strings.Contains(service.Name, "shuffle-backend") || strings.Contains(service.Name, "backend") || - strings.Contains(service.Name, "frontend") { + if skipCheckInCleanup(service.Name) { continue } @@ -244,6 +248,10 @@ func cleanupExistingNodes(ctx context.Context) error { } 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) From 9653de2a6a142dc9e33a43bb7e0584ca59d61387 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Tue, 9 Jul 2024 17:55:01 +0530 Subject: [PATCH 07/11] fix[k8s]: more cleanup fixes --- functions/onprem/orborus/orborus.go | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 4cfc603e..836db0ce 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -178,13 +178,12 @@ func getThisContainerId() { func skipCheckInCleanup(name string) bool { - - return strings.HasPrefix(name, "backend-") || - strings.HasPrefix(name, "shuffle-backend") || - strings.HasPrefix(name, "frontend-") || - strings.HasPrefix(name, "orborus-") || - strings.HasPrefix(name, "opensearch-") || - strings.HasPrefix(name, "memcached-") + return strings.HasPrefix(name, "backend") || + strings.HasPrefix(name, "shuffle-backend") || + strings.HasPrefix(name, "frontend") || + strings.HasPrefix(name, "orborus") || + strings.HasPrefix(name, "opensearch") || + strings.HasPrefix(name, "memcached") } func cleanupExistingNodes(ctx context.Context) error { From da76c43356fb026a50da8809b57191fb2ad6fa9c Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Tue, 9 Jul 2024 17:55:45 +0530 Subject: [PATCH 08/11] fix[k8s]: more cleanup fixes (1) --- functions/onprem/orborus/orborus.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 836db0ce..9ce03cf5 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -181,9 +181,13 @@ func skipCheckInCleanup(name string) bool { return strings.HasPrefix(name, "backend") || strings.HasPrefix(name, "shuffle-backend") || strings.HasPrefix(name, "frontend") || + strings.HasPrefix(name, "shuffle-frontend") || strings.HasPrefix(name, "orborus") || + strings.HasPrefix(name, "shuffle-orborus") || strings.HasPrefix(name, "opensearch") || + strings.HasPrefix(name, "shuffle-opensearch") || strings.HasPrefix(name, "memcached") + strings.HasPrefix(name, "shuffle-memcached") } func cleanupExistingNodes(ctx context.Context) error { From 26733bc3cb1dff71c6d35ccf67258208d3491356 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Tue, 9 Jul 2024 18:28:33 +0530 Subject: [PATCH 09/11] fix[k8s]: more cleanup fixes (2) --- functions/onprem/orborus/orborus.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 9ce03cf5..847cd29e 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -186,7 +186,7 @@ func skipCheckInCleanup(name string) bool { strings.HasPrefix(name, "shuffle-orborus") || strings.HasPrefix(name, "opensearch") || strings.HasPrefix(name, "shuffle-opensearch") || - strings.HasPrefix(name, "memcached") + strings.HasPrefix(name, "memcached") || strings.HasPrefix(name, "shuffle-memcached") } From 2f5d4130964c44f8a1e385f473707d957d5f9574 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Tue, 16 Jul 2024 18:59:39 +0530 Subject: [PATCH 10/11] feat: adding all-in-one.yaml --- functions/kubernetes/all-in-one.yaml | 122 ++++++++++++++++++++++----- 1 file changed, 100 insertions(+), 22 deletions(-) diff --git a/functions/kubernetes/all-in-one.yaml b/functions/kubernetes/all-in-one.yaml index 746215e9..9ac8dfaf 100644 --- a/functions/kubernetes/all-in-one.yaml +++ b/functions/kubernetes/all-in-one.yaml @@ -1,15 +1,27 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: shuffle + --- apiVersion: storage.k8s.io/v1 kind: StorageClass metadata: name: shuffle-data + namespace: shuffle provisioner: kubernetes.io/no-provisioner volumeBindingMode: WaitForFirstConsumer --- apiVersion: v1 +metadata: + namespace: shuffle + creationTimestamp: null + labels: + io.kompose.service: backend-env + name: env data: BACKEND_HOSTNAME: shuffle-backend BACKEND_PORT: "5001" @@ -51,11 +63,13 @@ data: SHUFFLE_OPENSEARCH_APIKEY: "" SHUFFLE_OPENSEARCH_CERTIFICATE_FILE: "" SHUFFLE_OPENSEARCH_CLOUDID: "" + KUBERNETES_NAMESPACE: shuffle SHUFFLE_OPENSEARCH_INDEX_PREFIX: "" SHUFFLE_OPENSEARCH_PASSWORD: admin SHUFFLE_OPENSEARCH_PROXY: "" SHUFFLE_OPENSEARCH_SKIPSSL_VERIFY: "true" SHUFFLE_OPENSEARCH_URL: https://opensearch:9200 + SHUFFLE_MEMCACHED: shuffle-memcached:11211 SHUFFLE_OPENSEARCH_USERNAME: admin SHUFFLE_ORBORUS_STARTUP_DELAY: "\t\t" SHUFFLE_PASS_APP_PROXY: "FALSE" @@ -68,11 +82,6 @@ data: REGISTRY_AUTH: "false" SHUFFLE_KUBERNETES_WORKER: "ghcr.io/shuffle/shuffle-worker:nightly" kind: ConfigMap -metadata: - creationTimestamp: null - labels: - io.kompose.service: backend-env - name: env --- @@ -110,7 +119,8 @@ roleRef: apiVersion: v1 kind: PersistentVolume metadata: - name: shuffle-os-pv + name: shuffle-os-pv + namespace: shuffle spec: capacity: storage: 10Gi # Adjust the storage size as per your requirements @@ -126,6 +136,7 @@ spec: apiVersion: v1 kind: PersistentVolumeClaim metadata: + namespace: shuffle creationTimestamp: null labels: io.kompose.service: opensearch-claim0 @@ -143,6 +154,7 @@ status: {} apiVersion: apps/v1 kind: Deployment metadata: + namespace: shuffle annotations: kompose.cmd: kompose convert -f docker-compose.yml kompose.version: 1.26.0 (40646f47) @@ -222,6 +234,7 @@ status: {} apiVersion: v1 kind: Service metadata: + namespace: shuffle annotations: kompose.cmd: kompose convert -f docker-compose.yml kompose.version: 1.26.0 (40646f47) @@ -244,7 +257,8 @@ status: apiVersion: v1 kind: PersistentVolume metadata: - name: shuffle-apps-pv + namespace: shuffle + name: shuffle-apps-pv spec: capacity: storage: 5Gi @@ -259,7 +273,8 @@ spec: apiVersion: v1 kind: PersistentVolume metadata: - name: shuffle-files-pv + namespace: shuffle + name: shuffle-files-pv spec: capacity: storage: 5Gi @@ -272,20 +287,21 @@ spec: --- - apiVersion: v1 - kind: PersistentVolumeClaim - metadata: - creationTimestamp: null - labels: - io.kompose.service: backend-files-claim - name: backend-files-claim - spec: - accessModes: - - ReadWriteOnce - storageClassName: shuffle-data - resources: - requests: - storage: 5Gi +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + namespace: shuffle + creationTimestamp: null + labels: + io.kompose.service: backend-files-claim + name: backend-files-claim +spec: + accessModes: + - ReadWriteOnce + storageClassName: shuffle-data + resources: + requests: + storage: 5Gi # status: {} --- @@ -293,6 +309,7 @@ spec: apiVersion: v1 kind: PersistentVolumeClaim metadata: + namespace: shuffle creationTimestamp: null labels: io.kompose.service: backend-apps-claim @@ -311,6 +328,48 @@ spec: apiVersion: apps/v1 kind: Deployment metadata: + name: shuffle-memcached + namespace: shuffle +spec: + replicas: 1 + selector: + matchLabels: + app: shuffle-memcached + template: + metadata: + labels: + app: shuffle-memcached + spec: + containers: + - name: shuffle-memcached + image: memcached:latest + ports: + - containerPort: 11211 + resources: {} + restartPolicy: Always + + +--- + +apiVersion: v1 +kind: Service +metadata: + namespace: shuffle + name: shuffle-memcached +spec: + ports: + - port: 11211 + targetPort: 11211 + selector: + app: shuffle-memcached + type: ClusterIP + +--- + +apiVersion: apps/v1 +kind: Deployment +metadata: + namespace: shuffle annotations: kompose.cmd: kompose convert -f docker-compose.yml kompose.version: 1.26.0 (40646f47) @@ -538,6 +597,11 @@ spec: configMapKeyRef: key: SHUFFLE_OPENSEARCH_APIKEY name: env + - name: SHUFFLE_MEMCACHED + valueFrom: + configMapKeyRef: + key: SHUFFLE_MEMCACHED + name: env - name: SHUFFLE_OPENSEARCH_CERTIFICATE_FILE valueFrom: configMapKeyRef: @@ -642,6 +706,7 @@ status: {} apiVersion: v1 kind: Service metadata: + namespace: shuffle annotations: kompose.cmd: kompose convert -f docker-compose.yml kompose.version: 1.26.0 (40646f47) @@ -664,6 +729,7 @@ status: apiVersion: apps/v1 kind: Deployment metadata: + namespace: shuffle annotations: kompose.cmd: kompose convert -f docker-compose.yml kompose.version: 1.26.0 (40646f47) @@ -705,6 +771,7 @@ status: {} apiVersion: v1 kind: Service metadata: + namespace: shuffle annotations: kompose.cmd: kompose convert -f docker-compose.yml kompose.version: 1.26.0 (40646f47) @@ -733,6 +800,7 @@ spec: apiVersion: apps/v1 kind: Deployment metadata: + namespace: shuffle annotations: kompose.cmd: kompose convert -f docker-compose.yml kompose.version: 1.26.0 (40646f47) @@ -779,6 +847,11 @@ spec: configMapKeyRef: key: IS_KUBERNETES name: env + - name: KUBERNETES_NAMESPACE + valueFrom: + configMapKeyRef: + key: KUBERNETES_NAMESPACE + name: env - name: REGISTRY_URL valueFrom: configMapKeyRef: @@ -790,6 +863,11 @@ spec: key: SHUFFLE_KUBERNETES_WORKER name: env + - name: SHUFFLE_MEMCACHED + valueFrom: + configMapKeyRef: + key: SHUFFLE_MEMCACHED + name: env image: ghcr.io/shuffle/shuffle-orborus:nightly #imagePullPolicy: Never name: shuffle-orborus From 0dd5913072b9ed0c14ca3487d7545f31cffc3dfc Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Tue, 16 Jul 2024 20:07:32 +0530 Subject: [PATCH 11/11] feat: making all-in-one.yaml work on GKE --- functions/kubernetes/all-in-one.yaml | 96 ++++++++++++++-------------- 1 file changed, 48 insertions(+), 48 deletions(-) diff --git a/functions/kubernetes/all-in-one.yaml b/functions/kubernetes/all-in-one.yaml index 9ac8dfaf..a428048f 100644 --- a/functions/kubernetes/all-in-one.yaml +++ b/functions/kubernetes/all-in-one.yaml @@ -116,22 +116,22 @@ roleRef: --- -apiVersion: v1 -kind: PersistentVolume -metadata: - name: shuffle-os-pv - namespace: shuffle -spec: - capacity: - storage: 10Gi # Adjust the storage size as per your requirements - accessModes: - - ReadWriteOnce # This allows read-write access to a single node - persistentVolumeReclaimPolicy: Retain # Adjust the reclaim policy as per your needs - storageClassName: shuffle-data # Set the desired storage class - hostPath: - path: /mnt/shuffle-data/open-search +# apiVersion: v1 +# kind: PersistentVolume +# metadata: +# name: shuffle-os-pv +# namespace: shuffle +# spec: +# capacity: +# storage: 10Gi # Adjust the storage size as per your requirements +# accessModes: +# - ReadWriteOnce # This allows read-write access to a single node +# persistentVolumeReclaimPolicy: Retain # Adjust the reclaim policy as per your needs +# storageClassName: standard-rwo # Set the desired storage class +# hostPath: +# path: /mnt/shuffle-data/open-search ---- +# --- apiVersion: v1 kind: PersistentVolumeClaim @@ -144,7 +144,7 @@ metadata: spec: accessModes: - ReadWriteOnce - storageClassName: shuffle-data + storageClassName: standard-rwo resources: requests: storage: 500Mi @@ -254,38 +254,38 @@ status: --- -apiVersion: v1 -kind: PersistentVolume -metadata: - namespace: shuffle - name: shuffle-apps-pv -spec: - capacity: - storage: 5Gi - accessModes: - - ReadWriteOnce - persistentVolumeReclaimPolicy: Retain - storageClassName: shuffle-data - hostPath: - path: /mnt/shuffle-data/backend +# apiVersion: v1 +# kind: PersistentVolume +# metadata: +# namespace: shuffle +# name: shuffle-apps-pv +# spec: +# capacity: +# storage: 5Gi +# accessModes: +# - ReadWriteOnce +# persistentVolumeReclaimPolicy: Retain +# storageClassName: shuffle-data +# hostPath: +# path: /mnt/shuffle-data/backend ---- -apiVersion: v1 -kind: PersistentVolume -metadata: - namespace: shuffle - name: shuffle-files-pv -spec: - capacity: - storage: 5Gi - accessModes: - - ReadWriteOnce - persistentVolumeReclaimPolicy: Retain - storageClassName: shuffle-data - hostPath: - path: /mnt/shuffle-data/backend +# --- +# apiVersion: v1 +# kind: PersistentVolume +# metadata: +# namespace: shuffle +# name: shuffle-files-pv +# spec: +# capacity: +# storage: 5Gi +# accessModes: +# - ReadWriteOnce +# persistentVolumeReclaimPolicy: Retain +# storageClassName: shuffle-data +# hostPath: +# path: /mnt/shuffle-data/backend ---- +# --- apiVersion: v1 kind: PersistentVolumeClaim @@ -298,7 +298,7 @@ metadata: spec: accessModes: - ReadWriteOnce - storageClassName: shuffle-data + storageClassName: standard-rwo resources: requests: storage: 5Gi @@ -317,7 +317,7 @@ metadata: spec: accessModes: - ReadWriteOnce - storageClassName: shuffle-data + storageClassName: standard-rwo resources: requests: storage: 5Gi