scheduler: configure restic job resources

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

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 {

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

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