diff --git a/plugins/scheduler-k3s/functions.go b/plugins/scheduler-k3s/functions.go index 7b95ddaa4..bec2feb88 100644 --- a/plugins/scheduler-k3s/functions.go +++ b/plugins/scheduler-k3s/functions.go @@ -8,18 +8,11 @@ import ( "time" "github.com/dokku/dokku/plugins/common" - batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/util/wait" - "k8s.io/client-go/kubernetes" corev1client "k8s.io/client-go/kubernetes/typed/core/v1" - "k8s.io/client-go/rest" - "k8s.io/client-go/tools/clientcmd" - clientcmdapi "k8s.io/client-go/tools/clientcmd/api" "k8s.io/client-go/tools/remotecommand" "k8s.io/kubectl/pkg/util/term" "k8s.io/kubernetes/pkg/client/conditions" @@ -35,12 +28,6 @@ type EnterPodInput struct { SelectedPod v1.Pod } -type GetPodInput struct { - Clientset KubernetesClient - Namespace string - Selector string -} - // StartCommandInput contains all the information needed to get the start command type StartCommandInput struct { // AppName is the name of the app @@ -61,12 +48,6 @@ type StartCommandOutput struct { Command []string } -type KubernetesClient struct { - Client kubernetes.Clientset - RestClient rest.Interface - RestConfig rest.Config -} - type WaitForPodBySelectorRunningInput struct { Clientset KubernetesClient Namespace string @@ -82,69 +63,18 @@ type WaitForPodToExistInput struct { Selector string } -func NewKubernetesClient() (KubernetesClient, error) { - clientConfig := KubernetesClientConfig() - restConf, err := clientConfig.ClientConfig() - if err != nil { - return KubernetesClient{}, err - } - - restConf.GroupVersion = &schema.GroupVersion{ - Group: "api", - Version: "v1", - } - - client, err := kubernetes.NewForConfig(restConf) - if err != nil { - return KubernetesClient{}, err - } - - restConf.NegotiatedSerializer = runtime.NewSimpleNegotiatedSerializer(runtime.SerializerInfo{}) - - restClient, err := rest.RESTClientFor(restConf) - if err != nil { - return KubernetesClient{}, err - } - - return KubernetesClient{ - Client: *client, - RestConfig: *restConf, - RestClient: restClient, - }, nil -} - -func KubernetesClientConfig() clientcmd.ClientConfig { - return clientcmd.NewNonInteractiveDeferredLoadingClientConfig( - &clientcmd.ClientConfigLoadingRules{ExplicitPath: KubeConfigPath}, - &clientcmd.ConfigOverrides{ClusterInfo: clientcmdapi.Cluster{Server: ""}}) -} - -func createKubernetesJob(ctx context.Context, job batchv1.Job) (batchv1.Job, error) { - clientset, err := NewKubernetesClient() - if err != nil { - return batchv1.Job{}, err - } - - createdJob, err := clientset.Client.BatchV1().Jobs(job.Namespace).Create(ctx, &job, metav1.CreateOptions{}) - if err != nil { - return batchv1.Job{}, err - } - - return *createdJob, nil -} - func createKubernetesNamespace(ctx context.Context, namespaceName string) error { clientset, err := NewKubernetesClient() if err != nil { return err } - namespaces, err := clientset.Client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{}) + namespaces, err := clientset.ListNamespaces(ctx) if err != nil { return err } - for _, namespace := range namespaces.Items { + for _, namespace := range namespaces { if namespace.Name == namespaceName { return nil } @@ -161,7 +91,9 @@ func createKubernetesNamespace(ctx context.Context, namespaceName string) error }, }, } - _, err = clientset.Client.CoreV1().Namespaces().Create(ctx, &namespace, metav1.CreateOptions{}) + _, err = clientset.CreateNamespace(ctx, CreateNamespaceInput{ + Namespace: namespace, + }) if err != nil { return err } @@ -318,17 +250,14 @@ func getStartCommand(input StartCommandInput) (StartCommandOutput, error) { }, nil } -func getPod(ctx context.Context, input GetPodInput) (v1.PodList, error) { - listOptions := metav1.ListOptions{LabelSelector: input.Selector} - podList, err := input.Clientset.Client.CoreV1().Pods(input.Namespace).List(ctx, listOptions) - return *podList, err -} - func isPodReady(ctx context.Context, clientset KubernetesClient, podName, namespace string) wait.ConditionWithContextFunc { return func(ctx context.Context) (bool, error) { fmt.Printf(".") // progress bar! - pod, err := clientset.Client.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) + pod, err := clientset.GetPod(ctx, GetPodInput{ + Name: podName, + Namespace: namespace, + }) if err != nil { return false, err } @@ -344,7 +273,7 @@ func isPodReady(ctx context.Context, clientset KubernetesClient, podName, namesp } func waitForPodBySelectorRunning(ctx context.Context, input WaitForPodBySelectorRunningInput) error { - podList, err := waitForPodToExist(ctx, WaitForPodToExistInput{ + pods, err := waitForPodToExist(ctx, WaitForPodToExistInput{ Clientset: input.Clientset, Namespace: input.Namespace, RetryCount: 3, @@ -354,29 +283,28 @@ func waitForPodBySelectorRunning(ctx context.Context, input WaitForPodBySelector return fmt.Errorf("Error waiting for pod to exist: %w", err) } - if len(podList.Items) == 0 { + if len(pods) == 0 { return fmt.Errorf("no pods in %s with selector %s", input.Namespace, input.Selector) } timeout := time.Duration(input.Timeout) * time.Second - for _, pod := range podList.Items { + for _, pod := range pods { if err := wait.PollUntilContextTimeout(ctx, time.Second, timeout, false, input.Waiter(ctx, input.Clientset, pod.Name, pod.Namespace)); err != nil { print("\n") - return err + return fmt.Errorf("Error waiting for pod to be ready: %w", err) } } print("\n") return nil } -func waitForPodToExist(ctx context.Context, input WaitForPodToExistInput) (v1.PodList, error) { - var podList v1.PodList +func waitForPodToExist(ctx context.Context, input WaitForPodToExistInput) ([]v1.Pod, error) { + var pods []v1.Pod var err error for i := 0; i < input.RetryCount; i++ { - podList, err = getPod(ctx, GetPodInput{ - Clientset: input.Clientset, - Namespace: input.Namespace, - Selector: input.Selector, + pods, err = input.Clientset.ListPods(ctx, ListPodsInput{ + Namespace: input.Namespace, + LabelSelector: input.Selector, }) if err == nil { break @@ -384,7 +312,7 @@ func waitForPodToExist(ctx context.Context, input WaitForPodToExistInput) (v1.Po time.Sleep(1 * time.Second) } if err != nil { - return podList, fmt.Errorf("Error listing pods: %w", err) + return pods, fmt.Errorf("Error listing pods: %w", err) } - return podList, nil + return pods, nil } diff --git a/plugins/scheduler-k3s/helm.go b/plugins/scheduler-k3s/helm.go index b846bf154..067bbffd6 100644 --- a/plugins/scheduler-k3s/helm.go +++ b/plugins/scheduler-k3s/helm.go @@ -17,7 +17,6 @@ import ( "helm.sh/helm/v3/pkg/chart/loader" "helm.sh/helm/v3/pkg/kube" "helm.sh/helm/v3/pkg/storage/driver" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) var DevNullPrinter = func(format string, v ...interface{}) {} @@ -106,7 +105,10 @@ func (h *HelmAgent) DeleteRevision(releaseName string, revision int) error { } secretName := fmt.Sprintf("sh.helm.release.v1.%s.v%d", releaseName, revision) - err = clientset.Client.CoreV1().Secrets(h.Namespace).Delete(context.Background(), secretName, metav1.DeleteOptions{}) + err = clientset.DeleteSecret(context.Background(), DeleteSecretInput{ + Name: secretName, + Namespace: h.Namespace, + }) if err != nil { return fmt.Errorf("Error deleting secret: %w", err) } diff --git a/plugins/scheduler-k3s/k8s.go b/plugins/scheduler-k3s/k8s.go new file mode 100644 index 000000000..6c35d2cb4 --- /dev/null +++ b/plugins/scheduler-k3s/k8s.go @@ -0,0 +1,257 @@ +package scheduler_k3s + +import ( + "context" + "errors" + + appsv1 "k8s.io/api/apps/v1" + autoscalingv1 "k8s.io/api/autoscaling/v1" + batchv1 "k8s.io/api/batch/v1" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + clientcmdapi "k8s.io/client-go/tools/clientcmd/api" + "k8s.io/utils/ptr" +) + +// KubernetesClient is a wrapper around the Kubernetes client +type KubernetesClient struct { + // Client is the Kubernetes client + Client kubernetes.Clientset + + // RestClient is the Kubernetes REST client + RestClient rest.Interface + + // RestConfig is the Kubernetes REST config + RestConfig rest.Config +} + +// NewKubernetesClient creates a new Kubernetes client +func NewKubernetesClient() (KubernetesClient, error) { + clientConfig := KubernetesClientConfig() + restConf, err := clientConfig.ClientConfig() + if err != nil { + return KubernetesClient{}, err + } + + restConf.GroupVersion = &schema.GroupVersion{ + Group: "api", + Version: "v1", + } + + client, err := kubernetes.NewForConfig(restConf) + if err != nil { + return KubernetesClient{}, err + } + + restConf.NegotiatedSerializer = runtime.NewSimpleNegotiatedSerializer(runtime.SerializerInfo{}) + + restClient, err := rest.RESTClientFor(restConf) + if err != nil { + return KubernetesClient{}, err + } + + return KubernetesClient{ + Client: *client, + RestConfig: *restConf, + RestClient: restClient, + }, nil +} + +// KubernetesClientConfig returns a Kubernetes client config +func KubernetesClientConfig() clientcmd.ClientConfig { + return clientcmd.NewNonInteractiveDeferredLoadingClientConfig( + &clientcmd.ClientConfigLoadingRules{ExplicitPath: KubeConfigPath}, + &clientcmd.ConfigOverrides{ClusterInfo: clientcmdapi.Cluster{Server: ""}}) +} + +// CreateJobInput contains all the information needed to create a Kubernetes job +type CreateJobInput struct { + // Job is the Kubernetes job + Job batchv1.Job + + // Namespace is the Kubernetes namespace + Namespace string +} + +// CreateJob creates a Kubernetes job +func (k KubernetesClient) CreateJob(ctx context.Context, input CreateJobInput) (batchv1.Job, error) { + job, err := k.Client.BatchV1().Jobs(input.Namespace).Create(ctx, &input.Job, metav1.CreateOptions{}) + if err != nil { + return batchv1.Job{}, err + } + + if job == nil { + return batchv1.Job{}, errors.New("job is nil") + } + + return *job, err +} + +// CreateNamespaceInput contains all the information needed to create a Kubernetes namespace +type CreateNamespaceInput struct { + // Namespace is the Kubernetes namespace + Namespace v1.Namespace +} + +// CreateNamespace creates a Kubernetes namespace +func (k KubernetesClient) CreateNamespace(ctx context.Context, input CreateNamespaceInput) (v1.Namespace, error) { + namespace, err := k.Client.CoreV1().Namespaces().Create(ctx, &input.Namespace, metav1.CreateOptions{}) + if err != nil { + return v1.Namespace{}, err + } + + if namespace == nil { + return v1.Namespace{}, errors.New("namespace is nil") + } + + return *namespace, err +} + +// DeleteJobInput contains all the information needed to delete a Kubernetes job +type DeleteJobInput struct { + // Name is the Kubernetes job name + Name string + + // Namespace is the Kubernetes namespace + Namespace string +} + +// DeleteJob deletes a Kubernetes job +func (k KubernetesClient) DeleteJob(ctx context.Context, input DeleteJobInput) error { + return k.Client.BatchV1().Jobs(input.Namespace).Delete(ctx, input.Name, metav1.DeleteOptions{ + PropagationPolicy: ptr.To(metav1.DeletePropagationForeground), + }) +} + +// DeleteSecretInput contains all the information needed to delete a Kubernetes secret +type DeleteSecretInput struct { + // Name is the Kubernetes secret name + Name string + + // Namespace is the Kubernetes namespace + Namespace string +} + +// DeleteSecret deletes a Kubernetes secret +func (k KubernetesClient) DeleteSecret(ctx context.Context, input DeleteSecretInput) error { + return k.Client.CoreV1().Secrets(input.Namespace).Delete(ctx, input.Name, metav1.DeleteOptions{}) +} + +// GetJobInput contains all the information needed to get a Kubernetes job +type GetPodInput struct { + // Name is the Kubernetes pod name + Name string + + // Namespace is the Kubernetes namespace + Namespace string +} + +// GetJob gets a Kubernetes job +func (k KubernetesClient) GetPod(ctx context.Context, input GetPodInput) (v1.Pod, error) { + pod, err := k.Client.CoreV1().Pods(input.Namespace).Get(ctx, input.Name, metav1.GetOptions{}) + if err != nil { + return v1.Pod{}, err + } + + if pod == nil { + return v1.Pod{}, errors.New("pod is nil") + } + + return *pod, err +} + +// ListDeploymentsInput contains all the information needed to list Kubernetes deployments +type ListDeploymentsInput struct { + // Namespace is the Kubernetes namespace + Namespace string + + // LabelSelector is the Kubernetes label selector + LabelSelector string +} + +// ListDeployments lists Kubernetes deployments +func (k KubernetesClient) ListDeployments(ctx context.Context, input ListDeploymentsInput) ([]appsv1.Deployment, error) { + listOptions := metav1.ListOptions{LabelSelector: input.LabelSelector} + deployments, err := k.Client.AppsV1().Deployments(input.Namespace).List(ctx, listOptions) + if err != nil { + return []appsv1.Deployment{}, err + } + + if deployments == nil { + return []appsv1.Deployment{}, errors.New("deployments is nil") + } + + return deployments.Items, nil +} + +// ListNamespaces lists Kubernetes namespaces +func (k KubernetesClient) ListNamespaces(ctx context.Context) ([]v1.Namespace, error) { + namespaces, err := k.Client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{}) + if err != nil { + return []v1.Namespace{}, err + } + if namespaces == nil { + return []v1.Namespace{}, errors.New("namespaces is nil") + } + + return namespaces.Items, nil +} + +// ListPodsInput contains all the information needed to list Kubernetes pods +type ListPodsInput struct { + // Namespace is the Kubernetes namespace + Namespace string + + // LabelSelector is the Kubernetes label selector + LabelSelector string +} + +// ListPods lists Kubernetes pods +func (k KubernetesClient) ListPods(ctx context.Context, input ListPodsInput) ([]v1.Pod, error) { + listOptions := metav1.ListOptions{LabelSelector: input.LabelSelector} + podList, err := k.Client.CoreV1().Pods(input.Namespace).List(ctx, listOptions) + if err != nil { + return []v1.Pod{}, err + } + + if podList == nil { + return []v1.Pod{}, errors.New("pod list is nil") + } + + return podList.Items, err +} + +// ScaleDeploymentInput contains all the information needed to scale a Kubernetes deployment +type ScaleDeploymentInput struct { + // Name is the Kubernetes deployment name + Name string + + // Namespace is the Kubernetes namespace + Namespace string + + // Replicas is the number of replicas to scale to + Replicas int32 +} + +// ScaleDeployment scales a Kubernetes deployment +func (k KubernetesClient) ScaleDeployment(ctx context.Context, input ScaleDeploymentInput) error { + _, err := k.Client.AppsV1().Deployments(input.Namespace).UpdateScale(ctx, input.Name, &autoscalingv1.Scale{ + ObjectMeta: metav1.ObjectMeta{ + Name: input.Name, + Namespace: input.Namespace, + }, + Spec: autoscalingv1.ScaleSpec{ + Replicas: input.Replicas, + }, + }, metav1.UpdateOptions{}) + if err != nil { + return err + } + + return nil +} diff --git a/plugins/scheduler-k3s/template.go b/plugins/scheduler-k3s/template.go index eb1a13778..1da582acc 100644 --- a/plugins/scheduler-k3s/template.go +++ b/plugins/scheduler-k3s/template.go @@ -1,6 +1,7 @@ package scheduler_k3s import ( + "crypto/rand" "fmt" "os" "path/filepath" @@ -534,12 +535,19 @@ func templateKubernetesJob(input Job) (batchv1.Job, error) { }) } + n := 5 + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + panic(err) + } + suffix := strings.ToLower(fmt.Sprintf("%X", b)) + job := batchv1.Job{ ObjectMeta: metav1.ObjectMeta{ - GenerateName: fmt.Sprintf("%s-%s-", input.AppName, input.ProcessType), - Namespace: input.Namespace, - Labels: labels, - Annotations: annotations, + Name: fmt.Sprintf("%s-%s-%s", input.AppName, input.ProcessType, suffix), + Namespace: input.Namespace, + Labels: labels, + Annotations: annotations, }, Spec: batchv1.JobSpec{ BackoffLimit: ptr.To(int32(0)), diff --git a/plugins/scheduler-k3s/triggers.go b/plugins/scheduler-k3s/triggers.go index 6b6447b11..f3f34713b 100644 --- a/plugins/scheduler-k3s/triggers.go +++ b/plugins/scheduler-k3s/triggers.go @@ -10,7 +10,6 @@ import ( "os" "os/signal" "path/filepath" - "slices" "strconv" "strings" "syscall" @@ -25,10 +24,8 @@ import ( orderedmap "github.com/wk8/go-ordered-map/v2" "gopkg.in/yaml.v3" appsv1 "k8s.io/api/apps/v1" - autoscalingv1 "k8s.io/api/autoscaling/v1" corev1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/kubernetes/pkg/client/conditions" "k8s.io/utils/ptr" ) @@ -444,41 +441,33 @@ func TriggerSchedulerEnter(scheduler string, appName string, processType string, labelSelector = append(labelSelector, fmt.Sprintf("dokku.com/process-type=%s", processType)) } - podList, err := clientset.Client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ + pods, err := clientset.ListPods(ctx, ListPodsInput{ + Namespace: namespace, LabelSelector: strings.Join(labelSelector, ","), }) if err != nil { return fmt.Errorf("Error listing pods: %w", err) } - if podList.Items == nil { + if len(pods) == 0 { return fmt.Errorf("No pods found for app %s", appName) } - if len(podList.Items) == 0 { - return fmt.Errorf("No pods found for app %s", appName) - } - - pods := []string{} - for _, pod := range podList.Items { - pods = append(pods, pod.Name) - } - slices.Sort(pods) processIndex-- - if processIndex > len(podList.Items) { + if processIndex > len(pods) { return fmt.Errorf("Process index %d out of range for app %s", processIndex, appName) } var selectedPod corev1.Pod if podName != "" { - for _, pod := range podList.Items { + for _, pod := range pods { if pod.Name == podName { selectedPod = pod break } } } else { - selectedPod = podList.Items[processIndex] + selectedPod = pods[processIndex] } processType, ok := selectedPod.Labels["dokku.com/process-type"] @@ -549,18 +538,18 @@ func TriggerSchedulerLogs(scheduler string, appName string, processType string, } namespace := common.PropertyGetDefault("scheduler-k3s", appName, "namespace", "default") - podList, err := clientset.Client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ + pods, err := clientset.ListPods(ctx, ListPodsInput{ + Namespace: namespace, LabelSelector: strings.Join(labelSelector, ","), }) if err != nil { return fmt.Errorf("Error listing pods: %w", err) } - if len(podList.Items) == 0 { + if len(pods) == 0 { return fmt.Errorf("No pods found for app %s", appName) } ch := make(chan bool) - podItems := podList.Items if os.Getenv("FORCE_TTY") == "1" { color.NoColor = false @@ -575,7 +564,7 @@ func TriggerSchedulerLogs(scheduler string, appName string, processType string, color.FgMagenta, } // colorIndex := 0 - for i := 0; i < len(podItems); i++ { + for i := 0; i < len(pods); i++ { if processIndex > 0 && i != (processIndex-1) { continue } @@ -589,7 +578,7 @@ func TriggerSchedulerLogs(scheduler string, appName string, processType string, podColor := colors[i%len(colors)] dynoText := color.New(podColor).SprintFunc() - podName := podItems[i].Name + podName := pods[i].Name podLogs, err := clientset.Client.CoreV1().Pods(namespace).GetLogs(podName, &logOptions).Stream(ctx) if err != nil { return err @@ -798,6 +787,11 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] color.NoColor = false } + clientset, err := NewKubernetesClient() + if err != nil { + return fmt.Errorf("Error creating kubernetes client: %w", err) + } + ctx, cancel := context.WithCancel(context.Background()) signals := make(chan os.Signal, 1) signal.Notify(signals, os.Interrupt, syscall.SIGHUP, @@ -806,21 +800,34 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] syscall.SIGTERM) go func() { <-signals + if attachToPod { + clientset.DeleteJob(ctx, DeleteJobInput{ // nolint: errcheck + Name: job.Name, + Namespace: namespace, + }) + } cancel() }() - createdJob, err := createKubernetesJob(ctx, job) + createdJob, err := clientset.CreateJob(ctx, CreateJobInput{ + Job: job, + Namespace: namespace, + }) if err != nil { return fmt.Errorf("Error creating job: %w", err) } - clientset, err := NewKubernetesClient() - if err != nil { - return fmt.Errorf("Error creating kubernetes client: %w", err) + if attachToPod { + defer func() { + clientset.DeleteJob(ctx, DeleteJobInput{ // nolint: errcheck + Name: job.Name, + Namespace: namespace, + }) + }() } batchJobSelector := fmt.Sprintf("batch.kubernetes.io/job-name=%s", createdJob.Name) - podList, err := waitForPodToExist(ctx, WaitForPodToExistInput{ + pods, err := waitForPodToExist(ctx, WaitForPodToExistInput{ Clientset: clientset, Namespace: namespace, RetryCount: 3, @@ -830,7 +837,7 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] return fmt.Errorf("Error waiting for pod to exist: %w", err) } - for _, pod := range podList.Items { + for _, pod := range pods { common.LogQuiet(pod.Name) } @@ -847,15 +854,14 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] }) if err != nil { if errors.Is(err, conditions.ErrPodCompleted) { - podList, podErr := getPod(ctx, GetPodInput{ - Clientset: clientset, - Namespace: namespace, - Selector: batchJobSelector, + pods, podErr := clientset.ListPods(ctx, ListPodsInput{ + Namespace: namespace, + LabelSelector: batchJobSelector, }) if podErr != nil { return fmt.Errorf("Error completed pod: %w", err) } - selectedPod := podList.Items[0] + selectedPod := pods[0] if selectedPod.Status.Phase == v1.PodFailed { for _, status := range selectedPod.Status.ContainerStatuses { if status.Name != fmt.Sprintf("%s-%s", appName, processType) { @@ -875,15 +881,14 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] return fmt.Errorf("Error waiting for pod to be running: %w", err) } - podList, err = getPod(ctx, GetPodInput{ - Clientset: clientset, - Namespace: namespace, - Selector: batchJobSelector, + pods, err = clientset.ListPods(ctx, ListPodsInput{ + Namespace: namespace, + LabelSelector: batchJobSelector, }) if err != nil { return fmt.Errorf("Error getting pod: %w", err) } - selectedPod := podList.Items[0] + selectedPod := pods[0] switch selectedPod.Status.Phase { case v1.PodFailed, v1.PodSucceeded: @@ -903,7 +908,7 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] return errors.New("Unable to attach as the pod has already exited with a successful exit code") } case v1.PodRunning: - err := enterPod(ctx, EnterPodInput{ + return enterPod(ctx, EnterPodInput{ AppName: appName, Clientset: clientset, Command: command, @@ -911,16 +916,6 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] ProcessType: processType, SelectedPod: selectedPod, }) - if err != nil { - return err - } - - err = clientset.Client.BatchV1().Jobs(namespace).Delete(ctx, createdJob.Name, metav1.DeleteOptions{ - PropagationPolicy: ptr.To(metav1.DeletePropagationForeground), - }) - if err != nil { - return fmt.Errorf("Error deleting pod: %w", err) - } return nil default: @@ -973,28 +968,25 @@ func TriggerSchedulerStop(scheduler string, appName string) error { }() namespace := common.PropertyGetDefault("scheduler-k3s", appName, "namespace", "default") - deploymentList, err := clientset.Client.AppsV1().Deployments(namespace).List(ctx, metav1.ListOptions{ + deployments, err := clientset.ListDeployments(ctx, ListDeploymentsInput{ + Namespace: namespace, LabelSelector: fmt.Sprintf("dokku.com/app-name=%s", appName), }) if err != nil { return fmt.Errorf("Error listing deployments: %w", err) } - for _, deployment := range deploymentList.Items { + for _, deployment := range deployments { processType, ok := deployment.Annotations["dokku.com/process-type"] if !ok { return fmt.Errorf("Deployment %s does not have a process type annotation", deployment.Name) } common.LogVerboseQuiet(fmt.Sprintf("Stopping %s process", processType)) - _, err := clientset.Client.AppsV1().Deployments(namespace).UpdateScale(ctx, deployment.Name, &autoscalingv1.Scale{ - ObjectMeta: metav1.ObjectMeta{ - Name: deployment.Name, - Namespace: namespace, - }, - Spec: autoscalingv1.ScaleSpec{ - Replicas: 0, - }, - }, metav1.UpdateOptions{}) + err := clientset.ScaleDeployment(ctx, ScaleDeploymentInput{ + Name: deployment.Name, + Namespace: namespace, + Replicas: 0, + }) if err != nil { return fmt.Errorf("Error updating deployment scale: %w", err) }