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

feat: implement cron jobs

Browse Source
This commit is contained in:
Jose Diaz-Gonzalez
2024-01-18 06:08:18 -05:00
parent 3f7e1d161a
commit 0127f836fa
3 changed files with 195 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/deployment/schedulers/k3s.md
View File

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

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

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

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

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