github/dokku
0
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor: move all k8s code behind client

Browse Source 
Also fix an issue where attaching to pods fails in rare cases
This commit is contained in:
Jose Diaz-Gonzalez
2024-01-18 16:14:33 -05:00
parent 1c5505f78b
commit 297853781f
5 changed files with 345 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
plugins/scheduler-k3s/functions.go
View File

@@ -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
}

6
plugins/scheduler-k3s/helm.go
View File

@@ -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)
}

257
plugins/scheduler-k3s/k8s.go Normal file
View File

@@ -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
}

16
plugins/scheduler-k3s/template.go
View File

@@ -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)),

112
plugins/scheduler-k3s/triggers.go
View File

@@ -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)
}