diff --git a/internal/config/config.go b/internal/config/config.go index 7039b05..4822f78 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -6,6 +6,8 @@ import ( "strconv" "strings" "time" + + "k8s.io/apimachinery/pkg/api/resource" ) const ( @@ -41,6 +43,10 @@ type Config struct { S3Region string JobTTLSeconds int32 JobNodeSelector map[string]string + JobCPURequest string + JobCPULimit string + JobMemoryRequest string + JobMemoryLimit string WorkerServiceAccount string ListenAddr string LonghornURL string @@ -98,6 +104,10 @@ func Load() (*Config, error) { cfg.WorkerServiceAccount = getenv("SOTERIA_JOB_SERVICE_ACCOUNT") cfg.ListenAddr = getenvDefault("SOTERIA_LISTEN_ADDR", defaultListenAddr) cfg.JobNodeSelector = parseNodeSelector(getenv("SOTERIA_JOB_NODE_SELECTOR")) + cfg.JobCPURequest = getenv("SOTERIA_JOB_CPU_REQUEST") + cfg.JobCPULimit = getenv("SOTERIA_JOB_CPU_LIMIT") + cfg.JobMemoryRequest = getenv("SOTERIA_JOB_MEMORY_REQUEST") + cfg.JobMemoryLimit = getenv("SOTERIA_JOB_MEMORY_LIMIT") cfg.LonghornURL = getenvDefault("SOTERIA_LONGHORN_URL", defaultLonghornURL) cfg.LonghornBackupMode = getenvDefault("SOTERIA_LONGHORN_BACKUP_MODE", defaultLonghornMode) cfg.AuthRequired = getenvBool("SOTERIA_AUTH_REQUIRED") @@ -172,6 +182,18 @@ func Load() (*Config, error) { if cfg.JobNodeSelector == nil { return nil, errors.New("SOTERIA_JOB_NODE_SELECTOR is invalid; expected key=value pairs") } + if err := validateQuantity("SOTERIA_JOB_CPU_REQUEST", cfg.JobCPURequest); err != nil { + return nil, err + } + if err := validateQuantity("SOTERIA_JOB_CPU_LIMIT", cfg.JobCPULimit); err != nil { + return nil, err + } + if err := validateQuantity("SOTERIA_JOB_MEMORY_REQUEST", cfg.JobMemoryRequest); err != nil { + return nil, err + } + if err := validateQuantity("SOTERIA_JOB_MEMORY_LIMIT", cfg.JobMemoryLimit); err != nil { + return nil, err + } if cfg.BackupDriver != "longhorn" && cfg.BackupDriver != "restic" { return nil, errors.New("SOTERIA_BACKUP_DRIVER must be longhorn or restic") } @@ -228,6 +250,16 @@ func getenv(key string) string { return strings.TrimSpace(os.Getenv(key)) } +func validateQuantity(key, raw string) error { + if strings.TrimSpace(raw) == "" { + return nil + } + if _, err := resource.ParseQuantity(raw); err != nil { + return errors.New(key + " must be a valid Kubernetes resource quantity") + } + return nil +} + func getenvDefault(key, value string) string { if v := getenv(key); v != "" { return v diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 26b89a4..599b189 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -111,6 +111,10 @@ func TestLoadSupportsResticAndB2Overrides(t *testing.T) { withEnv(t, "SOTERIA_S3_ENDPOINT", "https://b2.example.invalid") withEnv(t, "SOTERIA_S3_REGION", "us-west-000") withEnv(t, "SOTERIA_JOB_SERVICE_ACCOUNT", "soteria-worker") + withEnv(t, "SOTERIA_JOB_CPU_REQUEST", "100m") + withEnv(t, "SOTERIA_JOB_CPU_LIMIT", "1") + withEnv(t, "SOTERIA_JOB_MEMORY_REQUEST", "256Mi") + withEnv(t, "SOTERIA_JOB_MEMORY_LIMIT", "2Gi") withEnv(t, "SOTERIA_LISTEN_ADDR", ":9090") withEnv(t, "SOTERIA_JOB_NODE_SELECTOR", "hardware=rpi5,role=worker") withEnv(t, "SOTERIA_AUTH_REQUIRED", "true") @@ -155,6 +159,9 @@ func TestLoadSupportsResticAndB2Overrides(t *testing.T) { if cfg.WorkerServiceAccount != "soteria-worker" || cfg.ListenAddr != ":9090" { t.Fatalf("unexpected worker/listen config: %#v", cfg) } + if cfg.JobCPURequest != "100m" || cfg.JobCPULimit != "1" || cfg.JobMemoryRequest != "256Mi" || cfg.JobMemoryLimit != "2Gi" { + t.Fatalf("unexpected job resources: %#v", cfg) + } if cfg.JobNodeSelector["hardware"] != "rpi5" || cfg.JobNodeSelector["role"] != "worker" { t.Fatalf("unexpected node selector: %#v", cfg.JobNodeSelector) } @@ -297,6 +304,34 @@ func TestLoadRejectsInvalidConfigurations(t *testing.T) { }, substr: "SOTERIA_POLICY_BACKUPS_PER_CYCLE must be greater than or equal to zero", }, + { + name: "invalid job cpu request quantity", + env: map[string]string{ + "SOTERIA_JOB_CPU_REQUEST": "tiny-ish", + }, + substr: "SOTERIA_JOB_CPU_REQUEST must be a valid Kubernetes resource quantity", + }, + { + name: "invalid job cpu limit quantity", + env: map[string]string{ + "SOTERIA_JOB_CPU_LIMIT": "fast-ish", + }, + substr: "SOTERIA_JOB_CPU_LIMIT must be a valid Kubernetes resource quantity", + }, + { + name: "invalid job memory request quantity", + env: map[string]string{ + "SOTERIA_JOB_MEMORY_REQUEST": "big-ish", + }, + substr: "SOTERIA_JOB_MEMORY_REQUEST must be a valid Kubernetes resource quantity", + }, + { + name: "invalid job memory limit quantity", + env: map[string]string{ + "SOTERIA_JOB_MEMORY_LIMIT": "huge-ish", + }, + substr: "SOTERIA_JOB_MEMORY_LIMIT must be a valid Kubernetes resource quantity", + }, } for _, tc := range testCases { diff --git a/internal/k8s/job_manifests.go b/internal/k8s/job_manifests.go index e78a8f4..5a3019c 100644 --- a/internal/k8s/job_manifests.go +++ b/internal/k8s/job_manifests.go @@ -10,6 +10,7 @@ import ( batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -38,6 +39,7 @@ func buildBackupJob(cfg *config.Config, req api.BackupRequest, jobName, secretNa Command: []string{"/bin/sh", "-c"}, Args: []string{command}, Env: resticEnv(cfg, secretName, repository), + Resources: jobResources(cfg), VolumeMounts: []corev1.VolumeMount{ {Name: "data", MountPath: "/data", ReadOnly: true}, {Name: "cache", MountPath: "/cache"}, @@ -110,6 +112,7 @@ func buildRestoreJob(cfg *config.Config, req api.RestoreTestRequest, jobName, se Command: []string{"/bin/sh", "-c"}, Args: []string{command}, Env: resticEnv(cfg, secretName, repository), + Resources: jobResources(cfg), VolumeMounts: []corev1.VolumeMount{ {Name: "restore", MountPath: "/restore"}, {Name: "cache", MountPath: "/cache"}, @@ -168,6 +171,27 @@ func buildRestoreJob(cfg *config.Config, req api.RestoreTestRequest, jobName, se } } +func jobResources(cfg *config.Config) corev1.ResourceRequirements { + requests := corev1.ResourceList{} + limits := corev1.ResourceList{} + addQuantity(requests, corev1.ResourceCPU, cfg.JobCPURequest) + addQuantity(limits, corev1.ResourceCPU, cfg.JobCPULimit) + addQuantity(requests, corev1.ResourceMemory, cfg.JobMemoryRequest) + addQuantity(limits, corev1.ResourceMemory, cfg.JobMemoryLimit) + return corev1.ResourceRequirements{Requests: requests, Limits: limits} +} + +func addQuantity(items corev1.ResourceList, name corev1.ResourceName, raw string) { + if strings.TrimSpace(raw) == "" { + return + } + quantity, err := resource.ParseQuantity(raw) + if err != nil { + return + } + items[name] = quantity +} + func backupCommand(cfg *config.Config, req api.BackupRequest) string { mode := "on" if !dedupeEnabled(req.Dedupe) { diff --git a/internal/k8s/job_manifests_test.go b/internal/k8s/job_manifests_test.go index 7f175a6..957bd5f 100644 --- a/internal/k8s/job_manifests_test.go +++ b/internal/k8s/job_manifests_test.go @@ -6,6 +6,8 @@ import ( "scm.bstein.dev/bstein/soteria/internal/api" "scm.bstein.dev/bstein/soteria/internal/config" + + corev1 "k8s.io/api/core/v1" ) func TestBuildBackupJobAppliesSelectorsServiceAccountAndMetadata(t *testing.T) { @@ -18,6 +20,10 @@ func TestBuildBackupJobAppliesSelectorsServiceAccountAndMetadata(t *testing.T) { S3Region: "us-west-001", JobTTLSeconds: 3600, JobNodeSelector: map[string]string{"hardware": "rpi5"}, + JobCPURequest: "100m", + JobCPULimit: "1", + JobMemoryRequest: "256Mi", + JobMemoryLimit: "2Gi", WorkerServiceAccount: "soteria-worker", } keepLast := 3 @@ -52,6 +58,12 @@ func TestBuildBackupJobAppliesSelectorsServiceAccountAndMetadata(t *testing.T) { if len(container.Args) != 1 || !strings.Contains(container.Args[0], "restic unlock") || !strings.Contains(container.Args[0], "restic backup /data") || !strings.Contains(container.Args[0], "--keep-last 3") { t.Fatalf("expected backup command payload, got %#v", container.Args) } + if got := container.Resources.Requests[corev1.ResourceCPU]; got.String() != "100m" { + t.Fatalf("expected cpu request, got %s", got.String()) + } + if got := container.Resources.Limits[corev1.ResourceMemory]; got.String() != "2Gi" { + t.Fatalf("expected memory limit, got %s", got.String()) + } if len(job.Spec.Template.Spec.Volumes) != 2 || job.Spec.Template.Spec.Volumes[0].PersistentVolumeClaim == nil { t.Fatalf("expected pvc + cache volumes, got %#v", job.Spec.Template.Spec.Volumes) }