From 0127f836fa0932936e4809140846cc8c03aa7569 Mon Sep 17 00:00:00 2001 From: Jose Diaz-Gonzalez Date: Thu, 18 Jan 2024 06:08:18 -0500 Subject: [PATCH] feat: implement cron jobs --- docs/deployment/schedulers/k3s.md | 2 +- plugins/scheduler-k3s/template.go | 156 +++++++++++++++++++++++++++++- plugins/scheduler-k3s/triggers.go | 44 ++++++++- 3 files changed, 195 insertions(+), 7 deletions(-) diff --git a/docs/deployment/schedulers/k3s.md b/docs/deployment/schedulers/k3s.md index ced87c825..815474024 100644 --- a/docs/deployment/schedulers/k3s.md +++ b/docs/deployment/schedulers/k3s.md @@ -162,6 +162,7 @@ The following sections describe implemented and unimplemented scheduler function This plugin implements various functionality through `plugn` triggers to integrate with Docker for running apps on a single server. The following functionality is supported by the `scheduler-docker-local` plugin. - `apps:destroy` +- `cron` - `run` - The `scheduler-post-run` trigger is not always triggered - `run:detached` @@ -180,7 +181,6 @@ This plugin implements various functionality through `plugn` triggers to integra The following Dokku functionality is not implemented at this time. -- `cron` - `vector` log integration - persistent storage - healthchecks diff --git a/plugins/scheduler-k3s/template.go b/plugins/scheduler-k3s/template.go index 53c582ac4..2c425692d 100644 --- a/plugins/scheduler-k3s/template.go +++ b/plugins/scheduler-k3s/template.go @@ -82,6 +82,7 @@ type Job struct { Labels map[string]string Namespace string ProcessType string + Schedule string RemoveContainer bool WorkingDir string } @@ -163,6 +164,154 @@ func createIngressRoutesFiles(input CreateIngressRoutesInput) error { return nil } +func templateKubernetesCronJob(input Job) (batchv1.CronJob, error) { + if input.Schedule == "" { + return batchv1.CronJob{}, fmt.Errorf("Schedule cannot be empty") + } + 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/process-type": input.ProcessType, + } + annotations := map[string]string{ + "dokku.com/app-name": input.AppName, + "dokku.com/builder-type": input.ImageSourceType, + "dokku.com/deployment-id": "DEPLOYMENT_ID_QUOTED", + "dokku.com/managed": "true", + "dokku.com/process-type": input.ProcessType, + } + + for key, value := range input.Labels { + labels[key] = value + } + secretName := fmt.Sprintf("env-%s.DEPLOYMENT_ID", input.AppName) + + env := []corev1.EnvVar{} + for key, value := range input.Env { + env = append(env, corev1.EnvVar{ + Name: key, + Value: value, + }) + } + + job := batchv1.CronJob{ + ObjectMeta: metav1.ObjectMeta{ + GenerateName: fmt.Sprintf("%s-%s-", input.AppName, input.ProcessType), + Namespace: input.Namespace, + Labels: labels, + Annotations: annotations, + }, + Spec: batchv1.CronJobSpec{ + ConcurrencyPolicy: batchv1.AllowConcurrent, + FailedJobsHistoryLimit: ptr.To(int32(10)), + Schedule: input.Schedule, + StartingDeadlineSeconds: ptr.To(int64(60)), + SuccessfulJobsHistoryLimit: ptr.To(int32(10)), + Suspend: ptr.To(false), + TimeZone: ptr.To("Etc/UTC"), + JobTemplate: batchv1.JobTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: labels, + Annotations: annotations, + }, + Spec: batchv1.JobSpec{ + BackoffLimit: ptr.To(int32(0)), + PodReplacementPolicy: ptr.To(batchv1.Failed), + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Args: input.Command, + Name: fmt.Sprintf("%s-%s", input.AppName, input.ProcessType), + Env: env, + EnvFrom: []corev1.EnvFromSource{ + { + SecretRef: &corev1.SecretEnvSource{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: secretName, + }, + Optional: ptr.To(true), + }, + }, + }, + Image: input.Image, + ImagePullPolicy: corev1.PullAlways, + Resources: corev1.ResourceRequirements{ + Limits: corev1.ResourceList{}, + Requests: corev1.ResourceList{}, + }, + WorkingDir: input.WorkingDir, + }, + }, + RestartPolicy: corev1.RestartPolicyNever, + }, + }, + }, + }, + }, + } + + if input.Entrypoint != "" { + job.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Command = []string{input.Entrypoint} + } + + if input.RemoveContainer { + job.Spec.JobTemplate.Spec.TTLSecondsAfterFinished = ptr.To(int32(60)) + } + + if input.ImagePullSecrets != "" { + job.Spec.JobTemplate.Spec.Template.Spec.ImagePullSecrets = []corev1.LocalObjectReference{ + { + Name: input.ImagePullSecrets, + }, + } + } + + cpuLimit, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "limit", "cpu"}...) + if err != nil && cpuLimit != "" && cpuLimit != "0" { + cpuQuantity, err := resource.ParseQuantity(cpuLimit) + if err != nil { + return job, fmt.Errorf("Error parsing cpu limit: %w", err) + } + job.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Resources.Limits["cpu"] = cpuQuantity + } + nvidiaGpuLimit, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "limit", "nvidia-gpu"}...) + if err != nil && nvidiaGpuLimit != "" && nvidiaGpuLimit != "0" { + nvidiaGpuQuantity, err := resource.ParseQuantity(nvidiaGpuLimit) + if err != nil { + return job, fmt.Errorf("Error parsing nvidia-gpu limit: %w", err) + } + job.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Resources.Limits["nvidia.com/gpu"] = nvidiaGpuQuantity + } + memoryLimit, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "limit", "memory"}...) + if err != nil && memoryLimit != "" && memoryLimit != "0" { + memoryQuantity, err := resource.ParseQuantity(memoryLimit) + if err != nil { + return job, fmt.Errorf("Error parsing memory limit: %w", err) + } + job.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Resources.Limits["memory"] = memoryQuantity + } + + cpuRequest, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "reserve", "cpu"}...) + if err != nil && cpuRequest != "" && cpuRequest != "0" { + cpuQuantity, err := resource.ParseQuantity(cpuRequest) + if err != nil { + return job, fmt.Errorf("Error parsing cpu request: %w", err) + } + job.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Resources.Requests["cpu"] = cpuQuantity + } + memoryRequest, err := common.PlugnTriggerOutputAsString("resource-get-property", []string{input.AppName, input.ProcessType, "reserve", "memory"}...) + if err != nil && memoryRequest != "" && memoryRequest != "0" { + memoryQuantity, err := resource.ParseQuantity(memoryRequest) + if err != nil { + return job, fmt.Errorf("Error parsing memory request: %w", err) + } + job.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Resources.Requests["memory"] = memoryQuantity + } + + return job, nil +} + func templateKubernetesDeployment(input Deployment) (appsv1.Deployment, error) { labels := map[string]string{ "dokku.com/app-name": input.AppName, @@ -364,8 +513,6 @@ func templateKubernetesJob(input Job) (batchv1.Job, error) { "dokku.com/process-type": input.ProcessType, } - // todo: implement a controller to delete pods on completion - for key, value := range input.Labels { labels[key] = value } @@ -389,13 +536,16 @@ func templateKubernetesJob(input Job) (batchv1.Job, error) { Spec: batchv1.JobSpec{ BackoffLimit: ptr.To(int32(0)), Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: labels, + Annotations: annotations, + }, Spec: corev1.PodSpec{ Containers: []corev1.Container{ { Args: input.Command, Name: fmt.Sprintf("%s-%s", input.AppName, input.ProcessType), Env: env, - // todo: pull secret ref here EnvFrom: []corev1.EnvFromSource{ { SecretRef: &corev1.SecretEnvSource{ diff --git a/plugins/scheduler-k3s/triggers.go b/plugins/scheduler-k3s/triggers.go index c5eb25c04..412109562 100644 --- a/plugins/scheduler-k3s/triggers.go +++ b/plugins/scheduler-k3s/triggers.go @@ -18,7 +18,9 @@ import ( "github.com/dokku/dokku/plugins/common" "github.com/dokku/dokku/plugins/config" + "github.com/dokku/dokku/plugins/cron" "github.com/fatih/color" + "github.com/kballard/go-shellquote" "github.com/rancher/wharfie/pkg/registries" orderedmap "github.com/wk8/go-ordered-map/v2" "gopkg.in/yaml.v3" @@ -235,6 +237,7 @@ data: } } + workingDir := common.GetWorkingDir(appName, image) deployments := map[string]appsv1.Deployment{} i := 0 for processType := range processes { @@ -267,7 +270,7 @@ data: PortMaps: portMaps, ProcessType: processType, Replicas: replicaCountPlaceholder, - WorkingDir: common.GetWorkingDir(appName, image), + WorkingDir: workingDir, }) if err != nil { return fmt.Errorf("Error templating deployment: %w", err) @@ -280,11 +283,46 @@ data: Path: filepath.Join(chartDir, fmt.Sprintf("templates/deployment-%s.yaml", deployment.Name)), Replacements: replacements, }) - replacements.Delete(fmt.Sprintf("replicas: %d", replicaCountPlaceholder)) - if err != nil { return fmt.Errorf("Error printing deployment: %w", err) } + + replacements.Delete(fmt.Sprintf("replicas: %d", replicaCountPlaceholder)) + } + + cronEntries, err := cron.FetchCronEntries(appName) + if err != nil { + return fmt.Errorf("Error fetching cron entries: %w", err) + } + for _, cronEntry := range cronEntries { + words, err := shellquote.Split(cronEntry.Command) + if err != nil { + return fmt.Errorf("Error parsing cron command: %w", err) + } + cronJob, err := templateKubernetesCronJob(Job{ + AppName: appName, + Command: words, + Env: map[string]string{}, + Image: image, + ImagePullSecrets: imagePullSecrets, + ImageSourceType: imageSourceType, + Namespace: namespace, + ProcessType: cronEntry.ID, + Schedule: cronEntry.Schedule, + WorkingDir: workingDir, + }) + if err != nil { + return fmt.Errorf("Error templating cron job: %w", err) + } + + err = writeResourceToFile(WriteResourceInput{ + Object: &cronJob, + Path: filepath.Join(chartDir, fmt.Sprintf("templates/cron-job-%s.yaml", cronEntry.ID)), + Replacements: replacements, + }) + if err != nil { + return fmt.Errorf("Error printing cron job: %w", err) + } } if deployment, ok := deployments["web"]; ok {