diff --git a/plugins/scheduler-k3s/k8s.go b/plugins/scheduler-k3s/k8s.go index 6c35d2cb4..c3b5766e9 100644 --- a/plugins/scheduler-k3s/k8s.go +++ b/plugins/scheduler-k3s/k8s.go @@ -165,6 +165,34 @@ func (k KubernetesClient) GetPod(ctx context.Context, input GetPodInput) (v1.Pod return *pod, err } +// ListCronJobsInput contains all the information needed to list Kubernetes cron jobs +type ListCronJobsInput struct { + // LabelSelector is the Kubernetes label selector + LabelSelector string + + // Namespace is the Kubernetes namespace + Namespace string +} + +// ListCronJobs lists Kubernetes cron jobs +func (k KubernetesClient) ListCronJobs(ctx context.Context, input ListCronJobsInput) ([]batchv1.CronJob, error) { + listOptions := metav1.ListOptions{} + if input.LabelSelector != "" { + listOptions.LabelSelector = input.LabelSelector + } + + cronJobs, err := k.Client.BatchV1().CronJobs(input.Namespace).List(ctx, listOptions) + if err != nil { + return []batchv1.CronJob{}, err + } + + if cronJobs == nil { + return []batchv1.CronJob{}, errors.New("cron jobs is nil") + } + + return cronJobs.Items, err +} + // ListDeploymentsInput contains all the information needed to list Kubernetes deployments type ListDeploymentsInput struct { // Namespace is the Kubernetes namespace diff --git a/plugins/scheduler-k3s/template.go b/plugins/scheduler-k3s/template.go index 1da582acc..f5e63dc55 100644 --- a/plugins/scheduler-k3s/template.go +++ b/plugins/scheduler-k3s/template.go @@ -76,6 +76,7 @@ type Job struct { DeploymentID int64 Entrypoint string Env map[string]string + ID string Image string ImagePullSecrets string ImageSourceType string @@ -84,6 +85,7 @@ type Job struct { Namespace string ProcessType string Schedule string + Suffix string RemoveContainer bool WorkingDir string } @@ -172,11 +174,13 @@ func templateKubernetesCronJob(input Job) (batchv1.CronJob, error) { labels := map[string]string{ "dokku.com/app-name": input.AppName, "dokku.com/app-process-type": fmt.Sprintf("%s-%s", input.AppName, input.ProcessType), + "dokku.com/cron-id": input.ID, "dokku.com/process-type": input.ProcessType, } annotations := map[string]string{ "dokku.com/app-name": input.AppName, "dokku.com/builder-type": input.ImageSourceType, + "dokku.com/cron-id": input.ID, "dokku.com/deployment-id": "DEPLOYMENT_ID_QUOTED", "dokku.com/managed": "true", "dokku.com/process-type": input.ProcessType, @@ -195,12 +199,23 @@ func templateKubernetesCronJob(input Job) (batchv1.CronJob, error) { }) } + suffix := input.Suffix + if suffix == "" { + n := 5 + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + panic(err) + } + suffix = strings.ToLower(fmt.Sprintf("%X", b)) + } + annotations["dokku.com/job-suffix"] = suffix + job := batchv1.CronJob{ 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.CronJobSpec{ ConcurrencyPolicy: batchv1.AllowConcurrent, @@ -535,12 +550,16 @@ func templateKubernetesJob(input Job) (batchv1.Job, error) { }) } - n := 5 - b := make([]byte, n) - if _, err := rand.Read(b); err != nil { - panic(err) + suffix := input.Suffix + if suffix == "" { + n := 5 + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + panic(err) + } + suffix = strings.ToLower(fmt.Sprintf("%X", b)) } - suffix := strings.ToLower(fmt.Sprintf("%X", b)) + annotations["dokku.com/job-suffix"] = suffix job := batchv1.Job{ ObjectMeta: metav1.ObjectMeta{ diff --git a/plugins/scheduler-k3s/triggers.go b/plugins/scheduler-k3s/triggers.go index f3f34713b..a6a67f56b 100644 --- a/plugins/scheduler-k3s/triggers.go +++ b/plugins/scheduler-k3s/triggers.go @@ -276,7 +276,32 @@ data: if err != nil { return fmt.Errorf("Error fetching cron entries: %w", err) } + + clientset, err := NewKubernetesClient() + if err != nil { + return fmt.Errorf("Error creating kubernetes client: %w", err) + } + + cronJobs, err := clientset.ListCronJobs(ctx, ListCronJobsInput{ + LabelSelector: fmt.Sprintf("dokku.com/app-name=%s", appName), + Namespace: namespace, + }) + if err != nil { + return fmt.Errorf("Error listing cron jobs: %w", err) + } + for _, cronEntry := range cronEntries { + suffix := "" + for _, cronJob := range cronJobs { + if cronJob.Labels["dokku.com/cron-id"] == cronEntry.ID { + var ok bool + suffix, ok = cronJob.Annotations["dokku.com/job-suffix"] + if !ok { + suffix = "" + } + } + } + words, err := shellquote.Split(cronEntry.Command) if err != nil { return fmt.Errorf("Error parsing cron command: %w", err) @@ -285,12 +310,14 @@ data: AppName: appName, Command: words, Env: map[string]string{}, + ID: cronEntry.ID, Image: image, ImagePullSecrets: imagePullSecrets, ImageSourceType: imageSourceType, Namespace: namespace, - ProcessType: cronEntry.ID, + ProcessType: "cron", Schedule: cronEntry.Schedule, + Suffix: suffix, WorkingDir: workingDir, }) if err != nil { @@ -916,8 +943,6 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] ProcessType: processType, SelectedPod: selectedPod, }) - - return nil default: return fmt.Errorf("Unable to attach as the pod is in an unknown state: %s", selectedPod.Status.Phase) }