From c7fc95831fde06061c4488889ee898e4f558748a Mon Sep 17 00:00:00 2001 From: dhaval055 Date: Fri, 6 Oct 2023 16:38:25 +0000 Subject: [PATCH] added cleanup for apps and worker for k8s --- functions/k8s-confs/[stage-1 | 0 functions/k8s-confs/env-configmap.yaml | 2 +- functions/k8s-confs/orborus-deployment.yaml | 14 +- functions/k8s-confs/shuffle-registry.yaml | 68 +++++ functions/onprem/orborus/orborus.go | 113 +++++--- functions/onprem/worker/worker.go | 298 +++++++------------- 6 files changed, 255 insertions(+), 240 deletions(-) create mode 100644 functions/k8s-confs/[stage-1 create mode 100644 functions/k8s-confs/shuffle-registry.yaml diff --git a/functions/k8s-confs/[stage-1 b/functions/k8s-confs/[stage-1 new file mode 100644 index 00000000..e69de29b diff --git a/functions/k8s-confs/env-configmap.yaml b/functions/k8s-confs/env-configmap.yaml index ad02ad97..fb8d9d28 100644 --- a/functions/k8s-confs/env-configmap.yaml +++ b/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: diff --git a/functions/k8s-confs/orborus-deployment.yaml b/functions/k8s-confs/orborus-deployment.yaml index 7686c349..cc157289 100644 --- a/functions/k8s-confs/orborus-deployment.yaml +++ b/functions/k8s-confs/orborus-deployment.yaml @@ -24,10 +24,10 @@ spec: io.kompose.network/shuffle: "true" io.kompose.service: orborus spec: - volumes: - - name: docker-socket - hostPath: - path: /var/run/docker.sock + # volumes: + # - name: docker-socket + # hostPath: + # path: /var/run/docker.sock containers: - env: - name: BASE_URL @@ -61,9 +61,9 @@ spec: imagePullPolicy: Never name: shuffle-orborus resources: {} - volumeMounts: - - name: docker-socket - mountPath: /var/run/docker.sock + # volumeMounts: + # - name: docker-socket + # mountPath: /var/run/docker.sock hostname: shuffle-orborus restartPolicy: Always status: {} diff --git a/functions/k8s-confs/shuffle-registry.yaml b/functions/k8s-confs/shuffle-registry.yaml new file mode 100644 index 00000000..c6a80858 --- /dev/null +++ b/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 diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 47a79c12..adc66075 100755 --- a/functions/onprem/orborus/orborus.go +++ b/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{ - All: true, - }) + counter := 0 + if os.Getenv("IS_KUBERNETES") == "true" { + log.Printf("[INFO] getting running workers in kubernetes") - // Automatically updates the version - if err != nil { - log.Printf("[ERROR] Error getting containers: %s", err) + thresholdTime := time.Now().Add(time.Duration(-workerTimeout) * time.Second) - 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])) + clientset, err := getKubernetesClient() + if err != nil { + log.Printf("[ERROR] Failed getting kubernetes client: %s", err) + 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() - counter := 0 - 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 + for _, pod := range pods.Items { + if pod.Status.Phase == "Running" && pod.CreationTimestamp.Time.After(thresholdTime) { + counter++ + } + } + } else { + + containers, err := dockercli.ContainerList(ctx, types.ContainerListOptions{ + 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 } diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 15b7354f..fe47bfac 100755 --- a/functions/onprem/worker/worker.go +++ b/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) - for _, envStr := range env { - parts := strings.SplitN(envStr, "=", 2) - if len(parts) == 2 { - envMap[parts[0]] = parts[1] + namespace := "shuffle" + + envMap := make(map[string]string) + for _, envStr := range env { + parts := strings.SplitN(envStr, "=", 2) + if len(parts) == 2 { + envMap[parts[0]] = parts[1] + } } - } - clientset, err := getKubernetesClient() - if err != nil { - fmt.Println("[ERROR]Error getting kubernetes client:", err) - os.Exit(1) - } + clientset, err := getKubernetesClient() + if err != nil { + fmt.Println("[ERROR]Error getting kubernetes client:", err) + os.Exit(1) + } - log.Printf("[DEBUG] Got kubernetes client") - str := strings.ToLower(identifier) - strSplit := strings.Split(str, "_") - value := strSplit[0] - value = strings.ReplaceAll(value, "_", "-") + log.Printf("[DEBUG] Got kubernetes client") + str := strings.ToLower(identifier) + strSplit := strings.Split(str, "_") + value := strSplit[0] + value = strings.ReplaceAll(value, "_", "-") - //fix naming convention - podUuid := uuid.NewV4().String() - podName := fmt.Sprintf("%s-%s", value, podUuid) + //fix naming convention + podUuid := uuid.NewV4().String() + podName := fmt.Sprintf("%s-%s", value, podUuid) - pod := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{ - Name: podName, - Labels: map[string]string{ - "app": "shuffle-app", - }, - }, - Spec: corev1.PodSpec{ - Containers: []corev1.Container{ - { - Name: value, - Image: image, - Env: buildEnvVars(envMap), + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Labels: map[string]string{ + "app": "shuffle-app", + "executionId": workflowExecution.ExecutionId, + }, + }, + Spec: corev1.PodSpec{ + RestartPolicy: "Never", + Containers: []corev1.Container{ + { + 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 - //CPUShares: 128, - //CPUQuota: 10000, - //CPUPeriod: 100000, + return nil +} - // hostConfig := &container.HostConfig{ - // LogConfig: container.LogConfig{ - // Type: "json-file", - // Config: map[string]string{ - // "max-size": "10m", - // }, - // }, - // Resources: container.Resources{}, - // } +func cleanupExecution(clientset *kubernetes.Clientset, workflowExecution shuffle.WorkflowExecution, namespace string) error { - // 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 - // if cleanupEnv == "true" { - // hostConfig.AutoRemove = true - // } + podList, err := clientset.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{ + LabelSelector: labelSelector, + }) + if err != nil { + return fmt.Errorf("[ERROR]failed to list apps with label selector %s: %v", labelSelector, err) + } - // FIXME: Add proper foldermounts here - //log.Printf("\n\nPRE FOLDERMOUNT\n\n") - //volumeBinds := []string{"/tmp/shuffle-mount:/rules"} - //volumeBinds := []string{"/tmp/shuffle-mount:/rules"} - // volumeBinds := []string{} - // if len(volumeBinds) > 0 { - // 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 - // } + for _, pod := range podList.Items { + err := clientset.CoreV1().Pods(namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{}) + if err != nil { + return fmt.Errorf("failed to delete app %s: %v", pod.Name, err) + } + fmt.Printf("App %s in namespace %s deleted.\n", pod.Name, namespace) + } + podErr := clientset.CoreV1().Pods(namespace).Delete(context.TODO(), workerName, metav1.DeleteOptions{}) + if podErr != nil { + return fmt.Errorf("[ERROR] failed to delete the worker %s in namespace %s: %v", workerName, namespace, podErr) + } + 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)