diff --git a/backend/internal/api/routes.go b/backend/internal/api/routes.go index 7c41c30..459d09c 100644 --- a/backend/internal/api/routes.go +++ b/backend/internal/api/routes.go @@ -25,6 +25,7 @@ type API struct { getCredentials CredentialLookup checkInstanceAccess InstanceAccessCheck runner BenchmarkRunner + resourceDefaults coordinator.Resources store *RunStore lifecycleCtx context.Context runSlots chan struct{} @@ -33,7 +34,7 @@ type API struct { closing bool } -func NewAPI(getCredentials CredentialLookup, checkInstanceAccess InstanceAccessCheck, runner BenchmarkRunner, store *RunStore, lifecycleCtx context.Context) *API { +func NewAPI(getCredentials CredentialLookup, checkInstanceAccess InstanceAccessCheck, runner BenchmarkRunner, store *RunStore, lifecycleCtx context.Context, resourceDefaults coordinator.Resources) *API { if store == nil { store = NewRunStore() } @@ -44,6 +45,7 @@ func NewAPI(getCredentials CredentialLookup, checkInstanceAccess InstanceAccessC getCredentials: getCredentials, checkInstanceAccess: checkInstanceAccess, runner: runner, + resourceDefaults: resourceDefaults, store: store, lifecycleCtx: lifecycleCtx, runSlots: make(chan struct{}, maxConcurrentRuns), diff --git a/backend/internal/api/runs.go b/backend/internal/api/runs.go index 75acc71..9daaaaa 100644 --- a/backend/internal/api/runs.go +++ b/backend/internal/api/runs.go @@ -37,13 +37,14 @@ type Target struct { // CreateRunRequest contains the target and per-run benchmark settings. type CreateRunRequest struct { - Target Target `json:"target"` - Database string `json:"database"` - DurationSeconds int `json:"durationSeconds"` - Clients int `json:"clients"` - Threads int `json:"threads"` - Scale int `json:"scale"` - Initialize bool `json:"initialize"` + Target Target `json:"target"` + Database string `json:"database"` + DurationSeconds int `json:"durationSeconds"` + Clients int `json:"clients"` + Threads int `json:"threads"` + Scale int `json:"scale"` + Initialize bool `json:"initialize"` + Resources coordinator.Resources `json:"resources"` } // RunStatus describes the execution workflow. Running does not imply that @@ -52,27 +53,29 @@ type RunStatus string // Run is the record retained for status and result retrieval. type Run struct { - ID string `json:"id"` - Request CreateRunRequest `json:"request"` - Status RunStatus `json:"status"` - CreatedAt time.Time `json:"createdAt"` - CompletedAt *time.Time `json:"completedAt,omitempty"` - JobName string `json:"jobName,omitempty"` - Output string `json:"output,omitempty"` - OutputTruncated bool `json:"outputTruncated"` - Error string `json:"error,omitempty"` + ID string `json:"id"` + Request CreateRunRequest `json:"request"` + Resources coordinator.Resources `json:"resources"` + Status RunStatus `json:"status"` + CreatedAt time.Time `json:"createdAt"` + CompletedAt *time.Time `json:"completedAt,omitempty"` + JobName string `json:"jobName,omitempty"` + Output string `json:"output,omitempty"` + OutputTruncated bool `json:"outputTruncated"` + Error string `json:"error,omitempty"` } type RunStatusResponse struct { - ID string `json:"id"` - Status RunStatus `json:"status"` - Database string `json:"database,omitempty"` - CreatedAt time.Time `json:"createdAt"` - CompletedAt *time.Time `json:"completedAt,omitempty"` - JobName string `json:"jobName,omitempty"` - Output string `json:"output,omitempty"` - OutputTruncated bool `json:"outputTruncated"` - Error string `json:"error,omitempty"` + ID string `json:"id"` + Status RunStatus `json:"status"` + Database string `json:"database,omitempty"` + Resources coordinator.Resources `json:"resources"` + CreatedAt time.Time `json:"createdAt"` + CompletedAt *time.Time `json:"completedAt,omitempty"` + JobName string `json:"jobName,omitempty"` + Output string `json:"output,omitempty"` + OutputTruncated bool `json:"outputTruncated"` + Error string `json:"error,omitempty"` } func (a *API) createRun(w http.ResponseWriter, r *http.Request) { @@ -100,6 +103,12 @@ func (a *API) createRun(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, err.Error()) return } + options := optionsFromRequest(request) + resolved, err := coordinator.ResolveResources(a.resourceDefaults, options.Resources) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } if a.getCredentials == nil { writeError(w, http.StatusServiceUnavailable, "database credential lookup is unavailable") return @@ -151,20 +160,13 @@ func (a *API) createRun(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusInternalServerError, "failed to create benchmark run") return } - run, err := a.store.Create(runID, request) + run, err := a.store.Create(runID, request, resolved) if err != nil { <-a.runSlots a.runs.Done() writeError(w, http.StatusInternalServerError, "failed to create benchmark run") return } - options := coordinator.Options{ - Duration: request.DurationSeconds, - Clients: request.Clients, - Threads: request.Threads, - Scale: request.Scale, - Initialize: request.Initialize, - } runCtx, cancel := context.WithCancel(a.lifecycleCtx) go a.executeRun(runCtx, cancel, run.ID, connection, options) @@ -310,14 +312,18 @@ func validateCreateRunRequest(request CreateRunRequest) error { return errors.New("database must be a database name, not a connection string") } - options := coordinator.Options{ + return optionsFromRequest(request).Validate() +} + +func optionsFromRequest(request CreateRunRequest) coordinator.Options { + return coordinator.Options{ Duration: request.DurationSeconds, Clients: request.Clients, Threads: request.Threads, Scale: request.Scale, Initialize: request.Initialize, + Resources: request.Resources, } - return options.Validate() } var errUnsupportedContentType = errors.New("Content-Type must be application/json") @@ -402,6 +408,7 @@ func (a *API) getRun(w http.ResponseWriter, r *http.Request) { ID: run.ID, Status: run.Status, Database: run.Request.Database, + Resources: run.Resources, CreatedAt: run.CreatedAt, CompletedAt: run.CompletedAt, JobName: run.JobName, diff --git a/backend/internal/api/runs_test.go b/backend/internal/api/runs_test.go index 6880017..a0a0fbe 100644 --- a/backend/internal/api/runs_test.go +++ b/backend/internal/api/runs_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "io" "log" "net/http" @@ -40,7 +41,7 @@ func (f testRunnerFunc) Run(ctx context.Context, connection coordinator.Connecti func testAPI() *API { return NewAPI(func(context.Context, string, string, string, string) (*everest.Credentials, error) { return testCredentials(), nil - }, nil, testRunner{}, NewRunStore(), context.Background()) + }, nil, testRunner{}, NewRunStore(), context.Background(), coordinator.Resources{}) } func TestWriteErrorReturnsJSON(t *testing.T) { @@ -114,6 +115,7 @@ func TestCreateRunDecodesRequest(t *testing.T) { {"malformed", `{`, "application/json", http.StatusBadRequest}, {"null", `null`, "application/json", http.StatusBadRequest}, {"unknown field", `{"unexpected":true}`, "application/json", http.StatusBadRequest}, + {"unknown resource field", strings.TrimSuffix(validCreateRunBody, "}") + `,"resources":{"surprise":"1"}}`, "application/json", http.StatusBadRequest}, {"multiple values", `{} {}`, "application/json", http.StatusBadRequest}, {"wrong type", `[]`, "application/json", http.StatusBadRequest}, {"wrong content type", `{}`, "text/plain", http.StatusUnsupportedMediaType}, @@ -132,6 +134,123 @@ func TestCreateRunDecodesRequest(t *testing.T) { } } +func TestCreateRunResourceOverridesAndStatusSnapshot(t *testing.T) { + defaults := coordinator.Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + for _, test := range []struct { + name, resourceJSON string + overrides coordinator.Resources + resolved coordinator.Resources + fail bool + }{ + {"omitted resources", "", coordinator.Resources{}, defaults, false}, + {"partial overrides", `,"resources":{"cpuRequest":"500m","memoryLimit":"1Gi"}`, coordinator.Resources{CPURequest: "500m", MemoryLimit: "1Gi"}, coordinator.Resources{CPURequest: "500m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "1Gi"}, false}, + {"full overrides", `,"resources":{"cpuRequest":"500m","cpuLimit":"2","memoryRequest":"256Mi","memoryLimit":"1Gi"}`, coordinator.Resources{CPURequest: "500m", CPULimit: "2", MemoryRequest: "256Mi", MemoryLimit: "1Gi"}, coordinator.Resources{CPURequest: "500m", CPULimit: "2", MemoryRequest: "256Mi", MemoryLimit: "1Gi"}, false}, + {"failed run", `,"resources":{"cpuRequest":"250m"}`, coordinator.Resources{CPURequest: "250m"}, coordinator.Resources{CPURequest: "250m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"}, true}, + } { + t.Run(test.name, func(t *testing.T) { + store := NewRunStore() + started := make(chan coordinator.Options, 1) + runner := testRunnerFunc(func(_ context.Context, _ coordinator.Connection, options coordinator.Options) (coordinator.Result, error) { + started <- options + if test.fail { + return coordinator.Result{}, errors.New("runner failed") + } + return coordinator.Result{}, nil + }) + api := NewAPI(func(context.Context, string, string, string, string) (*everest.Credentials, error) { + return testCredentials(), nil + }, func(context.Context, string, string, string, string) error { return nil }, runner, store, context.Background(), defaults) + body := strings.TrimSuffix(validCreateRunBody, "}") + test.resourceJSON + "}" + r := httptest.NewRequest(http.MethodPost, "/api/runs", strings.NewReader(body)) + r.Header.Set("Authorization", "Bearer test-token") + r.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + api.createRun(w, r) + if w.Code != http.StatusAccepted { + t.Fatalf("create status=%d; body=%s", w.Code, w.Body.String()) + } + var accepted struct { + ID string `json:"id"` + } + if err := json.Unmarshal(w.Body.Bytes(), &accepted); err != nil || accepted.ID == "" { + t.Fatalf("acceptance response=%s; err=%v", w.Body.String(), err) + } + select { + case options := <-started: + if options.Resources != test.overrides { + t.Fatalf("runner overrides=%+v; want %+v", options.Resources, test.overrides) + } + case <-time.After(time.Second): + t.Fatal("runner was not started") + } + wantStatus := RunStatusSucceeded + if test.fail { + wantStatus = RunStatusFailed + } + deadline := time.Now().Add(time.Second) + for { + run, err := store.Get(accepted.ID) + if err == nil && run.Status == wantStatus { + if run.Resources != test.resolved || run.Request.Resources != test.overrides { + t.Fatalf("stored resources=%+v request=%+v; want %+v and %+v", run.Resources, run.Request.Resources, test.resolved, test.overrides) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("run did not complete: %+v, %v", run, err) + } + time.Sleep(time.Millisecond) + } + // A later change to deployment defaults must not rewrite stored history. + api.resourceDefaults = coordinator.Resources{CPURequest: "900m"} + statusRequest := httptest.NewRequest(http.MethodGet, "/api/runs/"+accepted.ID, nil) + statusRequest.Header.Set("Authorization", "Bearer test-token") + statusRequest.SetPathValue("id", accepted.ID) + statusWriter := httptest.NewRecorder() + api.getRun(statusWriter, statusRequest) + if statusWriter.Code != http.StatusOK { + t.Fatalf("status=%d; body=%s", statusWriter.Code, statusWriter.Body.String()) + } + var response RunStatusResponse + if err := json.Unmarshal(statusWriter.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Status != wantStatus || response.Resources != test.resolved { + t.Fatalf("status response=%+v; want status %s and resources %+v", response, wantStatus, test.resolved) + } + }) + } +} + +func TestCreateRunRejectsInvalidResourcesBeforeCredentialLookup(t *testing.T) { + defaults := coordinator.Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + for _, test := range []struct{ name, resources string }{ + {"invalid cpu", `{"cpuRequest":"bad"}`}, + {"zero limit", `{"memoryLimit":"0"}`}, + {"negative request", `{"cpuRequest":"-1"}`}, + {"cpu request exceeds default limit", `{"cpuRequest":"2"}`}, + {"memory limit below default request", `{"memoryLimit":"64Mi"}`}, + } { + t.Run(test.name, func(t *testing.T) { + lookupCalls := 0 + store := NewRunStore() + api := NewAPI(func(context.Context, string, string, string, string) (*everest.Credentials, error) { + lookupCalls++ + return testCredentials(), nil + }, nil, testRunner{}, store, context.Background(), defaults) + body := strings.TrimSuffix(validCreateRunBody, "}") + `,"resources":` + test.resources + "}" + r := httptest.NewRequest(http.MethodPost, "/api/runs", strings.NewReader(body)) + r.Header.Set("Authorization", "Bearer test-token") + r.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + api.createRun(w, r) + if w.Code != http.StatusBadRequest || lookupCalls != 0 || len(store.runs) != 0 || len(api.runSlots) != 0 { + t.Fatalf("invalid resource handling: status=%d lookups=%d runs=%d slots=%d body=%s", w.Code, lookupCalls, len(store.runs), len(api.runSlots), w.Body.String()) + } + }) + } +} + func TestCreateRunResolvesTargetCredentials(t *testing.T) { called := false api := NewAPI(func(ctx context.Context, token, cluster, namespace, instance string) (*everest.Credentials, error) { @@ -140,7 +259,7 @@ func TestCreateRunResolvesTargetCredentials(t *testing.T) { t.Fatalf("unexpected credential lookup arguments: token=%q target=%q/%q/%q", token, cluster, namespace, instance) } return testCredentials(), nil - }, nil, testRunner{}, NewRunStore(), context.Background()) + }, nil, testRunner{}, NewRunStore(), context.Background(), coordinator.Resources{}) r := httptest.NewRequest(http.MethodPost, "/api/runs", strings.NewReader(validCreateRunBody)) r.Header.Set("Authorization", "Bearer test-token") r.Header.Set("Content-Type", "application/json") @@ -168,7 +287,7 @@ func TestCreateRunStoresAndExecutesAsynchronously(t *testing.T) { }) api := NewAPI(func(context.Context, string, string, string, string) (*everest.Credentials, error) { return testCredentials(), nil - }, nil, runner, store, context.Background()) + }, nil, runner, store, context.Background(), coordinator.Resources{}) r := httptest.NewRequest(http.MethodPost, "/api/runs", strings.NewReader(validCreateRunBody)) r.Header.Set("Authorization", "Bearer test-token") r.Header.Set("Content-Type", "application/json") @@ -220,7 +339,7 @@ func TestCreateRunDefaultsToInstanceDatabase(t *testing.T) { }, nil, testRunnerFunc(func(_ context.Context, connection coordinator.Connection, _ coordinator.Options) (coordinator.Result, error) { databases <- connection.Database return coordinator.Result{}, nil - }), store, context.Background()) + }), store, context.Background(), coordinator.Resources{}) body := `{"target":{"k8sCluster":"local","namespace":"dbs","instance":"postgres-1"},"database":" ","durationSeconds":30,"clients":1,"threads":1,"scale":1}` r := httptest.NewRequest(http.MethodPost, "/api/runs", strings.NewReader(body)) r.Header.Set("Authorization", "Bearer test-token") @@ -259,7 +378,7 @@ func TestShutdownWaitsForActiveRuns(t *testing.T) { close(started) <-release return coordinator.Result{}, nil - }), NewRunStore(), context.Background()) + }), NewRunStore(), context.Background(), coordinator.Resources{}) r := httptest.NewRequest(http.MethodPost, "/api/runs", strings.NewReader(validCreateRunBody)) r.Header.Set("Authorization", "Bearer test-token") @@ -327,7 +446,7 @@ func TestCreateRunMapsCredentialLookupErrors(t *testing.T) { t.Run(test.name, func(t *testing.T) { api := NewAPI(func(context.Context, string, string, string, string) (*everest.Credentials, error) { return nil, test.err - }, nil, testRunner{}, NewRunStore(), context.Background()) + }, nil, testRunner{}, NewRunStore(), context.Background(), coordinator.Resources{}) r := httptest.NewRequest(http.MethodPost, "/api/runs", strings.NewReader(validCreateRunBody)) r.Header.Set("Authorization", "Bearer test-token") r.Header.Set("Content-Type", "application/json") @@ -417,11 +536,12 @@ func TestGetRun(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { store := NewRunStore() + snapshot := coordinator.Resources{CPURequest: "250m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} if test.exists { run, err := store.Create("run-123", CreateRunRequest{ Target: Target{K8sCluster: "local", Namespace: "databases", Instance: "postgres-1"}, Database: "app", - }) + }, snapshot) if err != nil { t.Fatal(err) } @@ -443,7 +563,7 @@ func TestGetRun(t *testing.T) { t.Fatalf("unexpected instance access check arguments: token=%q target=%q/%q/%q", token, cluster, namespace, instance) } return test.accessErr - }, testRunner{}, store, context.Background()) + }, testRunner{}, store, context.Background(), coordinator.Resources{}) if test.omitAccessCheck { api.checkInstanceAccess = nil } @@ -490,7 +610,7 @@ func TestGetRun(t *testing.T) { if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { t.Fatal(err) } - if response.ID != "run-123" || response.Status != test.status || response.Database != "app" || response.CreatedAt.IsZero() { + if response.ID != "run-123" || response.Status != test.status || response.Database != "app" || response.Resources != snapshot || response.CreatedAt.IsZero() { t.Fatalf("unexpected run status response: %+v", response) } if test.status == RunStatusRunning { diff --git a/backend/internal/api/store.go b/backend/internal/api/store.go index f4eee9c..3fcb349 100644 --- a/backend/internal/api/store.go +++ b/backend/internal/api/store.go @@ -24,7 +24,7 @@ type RunStore struct { } // Create stores a new running record. -func (s *RunStore) Create(id string, request CreateRunRequest) (Run, error) { +func (s *RunStore) Create(id string, request CreateRunRequest, resources coordinator.Resources) (Run, error) { if strings.TrimSpace(id) == "" { return Run{}, errors.New("run ID is required") } @@ -35,7 +35,7 @@ func (s *RunStore) Create(id string, request CreateRunRequest) (Run, error) { if _, exists := s.runs[id]; exists { return Run{}, ErrRunExists } - run := Run{ID: id, Request: request, Status: RunStatusRunning, CreatedAt: now.UTC()} + run := Run{ID: id, Request: request, Resources: resources, Status: RunStatusRunning, CreatedAt: now.UTC()} if s.runs == nil { s.runs = make(map[string]Run) } diff --git a/backend/internal/api/store_test.go b/backend/internal/api/store_test.go index d9711f0..3aa416f 100644 --- a/backend/internal/api/store_test.go +++ b/backend/internal/api/store_test.go @@ -13,13 +13,16 @@ func TestRunStoreLifecycle(t *testing.T) { for _, status := range []RunStatus{RunStatusSucceeded, RunStatusFailed} { t.Run(string(status), func(t *testing.T) { s := NewRunStore() - request := CreateRunRequest{Database: "benchmark"} - created, err := s.Create("run-1", request) + request := CreateRunRequest{Database: "benchmark", Resources: coordinator.Resources{CPURequest: "500m"}} + resolved := coordinator.Resources{CPURequest: "500m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + created, err := s.Create("run-1", request, resolved) if err != nil || created.Status != RunStatusRunning || created.CreatedAt.IsZero() { t.Fatalf("Create = %+v, %v", created, err) } created.Request.Database = "changed" - if _, err := s.Create("run-1", CreateRunRequest{}); !errors.Is(err, ErrRunExists) { + created.Request.Resources.CPURequest = "changed" + created.Resources.CPURequest = "changed" + if _, err := s.Create("run-1", CreateRunRequest{}, coordinator.Resources{}); !errors.Is(err, ErrRunExists) { t.Fatalf("duplicate Create: %v", err) } message := "" @@ -31,13 +34,15 @@ func TestRunStoreLifecycle(t *testing.T) { t.Fatal(err) } got, err := s.Get("run-1") - if err != nil || got.Request != request || got.Status != status || got.Output != result.Output || got.JobName != result.JobName || !got.OutputTruncated || got.Error != message || got.CompletedAt == nil { + if err != nil || got.Request != request || got.Resources != resolved || got.Status != status || got.Output != result.Output || got.JobName != result.JobName || !got.OutputTruncated || got.Error != message || got.CompletedAt == nil { t.Fatalf("Get = %+v, %v", got, err) } + got.Resources.MemoryLimit = "changed" + got.Request.Resources.CPURequest = "changed" *got.CompletedAt = time.Time{} again, _ := s.Get("run-1") - if again.CompletedAt.IsZero() { - t.Fatal("Get exposed stored timestamp pointer") + if again.CompletedAt.IsZero() || again.Resources != resolved || again.Request != request { + t.Fatalf("Get exposed stored values: %+v", again) } if err := s.Complete("run-1", RunStatusSucceeded, coordinator.Result{}, ""); !errors.Is(err, ErrRunAlreadyCompleted) { t.Fatalf("repeated Complete: %v", err) @@ -48,7 +53,7 @@ func TestRunStoreLifecycle(t *testing.T) { func TestRunStoreRejectsInvalidOperations(t *testing.T) { s := NewRunStore() - if _, err := s.Create(" ", CreateRunRequest{}); err == nil { + if _, err := s.Create(" ", CreateRunRequest{}, coordinator.Resources{}); err == nil { t.Fatal("accepted empty ID") } if _, err := s.Get("missing"); !errors.Is(err, ErrRunNotFound) { @@ -57,7 +62,7 @@ func TestRunStoreRejectsInvalidOperations(t *testing.T) { if err := s.Complete("missing", RunStatusSucceeded, coordinator.Result{}, ""); !errors.Is(err, ErrRunNotFound) { t.Fatal(err) } - if _, err := s.Create("run", CreateRunRequest{}); err != nil { + if _, err := s.Create("run", CreateRunRequest{}, coordinator.Resources{}); err != nil { t.Fatal(err) } for _, test := range []struct { @@ -78,7 +83,7 @@ func TestRunStoreRejectsInvalidOperations(t *testing.T) { func TestRunStoreConcurrentCompletion(t *testing.T) { s := NewRunStore() - if _, err := s.Create("run", CreateRunRequest{}); err != nil { + if _, err := s.Create("run", CreateRunRequest{}, coordinator.Resources{}); err != nil { t.Fatal(err) } var wg sync.WaitGroup @@ -119,7 +124,7 @@ func TestRunStoreExpiresCompletedRuns(t *testing.T) { s := NewRunStore() s.now = func() time.Time { return now } - if _, err := s.Create("completed", CreateRunRequest{}); err != nil { + if _, err := s.Create("completed", CreateRunRequest{}, coordinator.Resources{}); err != nil { t.Fatal(err) } if err := s.Complete("completed", test.status, coordinator.Result{}, test.message); err != nil { @@ -130,7 +135,7 @@ func TestRunStoreExpiresCompletedRuns(t *testing.T) { if _, err := s.Get("completed"); !errors.Is(err, ErrRunNotFound) { t.Fatalf("Get expired run error = %v, want %v", err, ErrRunNotFound) } - if _, err := s.Create("completed", CreateRunRequest{}); err != nil { + if _, err := s.Create("completed", CreateRunRequest{}, coordinator.Resources{}); err != nil { t.Fatalf("Create should prune the expired run before reusing its ID: %v", err) } }) @@ -142,11 +147,11 @@ func TestRunStoreRetainsRunningRunsPastRetention(t *testing.T) { s := NewRunStore() s.now = func() time.Time { return now } - if _, err := s.Create("running", CreateRunRequest{}); err != nil { + if _, err := s.Create("running", CreateRunRequest{}, coordinator.Resources{}); err != nil { t.Fatal(err) } now = now.Add(completedRunRetention + time.Second) - if _, err := s.Create("another", CreateRunRequest{}); err != nil { + if _, err := s.Create("another", CreateRunRequest{}, coordinator.Resources{}); err != nil { t.Fatal(err) } diff --git a/backend/internal/coordinator/coordinator.go b/backend/internal/coordinator/coordinator.go index 5939488..ca494ae 100644 --- a/backend/internal/coordinator/coordinator.go +++ b/backend/internal/coordinator/coordinator.go @@ -46,14 +46,6 @@ func (c Config) Validate() error { return nil } -// Resources uses Kubernetes quantity strings; empty fields remain unspecified. -type Resources struct { - CPURequest string - MemoryRequest string - CPULimit string - MemoryLimit string -} - // Coordinator holds shared configuration, not per-run credentials or options. // The Kubernetes client is injected so the coordinator remains testable. type Coordinator struct { diff --git a/backend/internal/coordinator/jobs.go b/backend/internal/coordinator/jobs.go index 88df241..9cb5576 100644 --- a/backend/internal/coordinator/jobs.go +++ b/backend/internal/coordinator/jobs.go @@ -14,7 +14,6 @@ import ( batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" @@ -247,7 +246,9 @@ func (c *Coordinator) createBenchmarkResources(ctx context.Context, connection C if err := options.Validate(); err != nil { return resources, fmt.Errorf("invalid benchmark options: %w", err) } - if _, err := resourceRequirements(c.config.Resources); err != nil { + // Validate the combined configuration before writing the credential Secret. + resolvedResources, err := ResolveResources(c.config.Resources, options.Resources) + if err != nil { return resources, fmt.Errorf("invalid runner resources: %w", err) } if err := ctx.Err(); err != nil { @@ -259,7 +260,7 @@ func (c *Coordinator) createBenchmarkResources(ctx context.Context, connection C return resources, err } - job, err := c.createBenchmarkJob(ctx, secret, options, labels) + job, err := c.createBenchmarkJob(ctx, secret, options, resolvedResources, labels) if err == nil { return benchmarkResources{secret: secret, job: job}, nil } @@ -367,8 +368,9 @@ func (c *Coordinator) cleanupBenchmarkResourcesWithContext(ctx context.Context, } // createBenchmarkJob builds and creates one runner Job in the configured -// workload namespace. It does not wait for completion or read logs. -func (c *Coordinator) createBenchmarkJob(ctx context.Context, secretRef SecretRef, options Options, labels map[string]string) (JobRef, error) { +// workload namespace. resolvedResources comes from the preflight check in +// createBenchmarkResources. It does not wait for completion or read logs. +func (c *Coordinator) createBenchmarkJob(ctx context.Context, secretRef SecretRef, options Options, resolvedResources Resources, labels map[string]string) (JobRef, error) { var reference JobRef if c == nil { return reference, errors.New("coordinator is nil") @@ -396,7 +398,7 @@ func (c *Coordinator) createBenchmarkJob(ctx context.Context, secretRef SecretRe if err != nil { return reference, err } - job, err := newBenchmarkJob(c.config, name, secretRef.Name, options, labels) + job, err := newBenchmarkJob(c.config, name, secretRef.Name, options, resolvedResources, labels) if err != nil { return reference, fmt.Errorf("build benchmark Job: %w", err) } @@ -418,7 +420,9 @@ func generateBenchmarkJobName() (string, error) { // newBenchmarkJob builds the per-run Job object. It only creates an in-memory // Kubernetes object; the caller is responsible for API calls and cleanup. -func newBenchmarkJob(config Config, name, secretName string, options Options, labels map[string]string) (*batchv1.Job, error) { +// resolvedResources must have been checked against deployment defaults before +// creating the credential Secret. +func newBenchmarkJob(config Config, name, secretName string, options Options, resolvedResources Resources, labels map[string]string) (*batchv1.Job, error) { if err := config.Validate(); err != nil { return nil, fmt.Errorf("invalid coordinator configuration: %w", err) } @@ -431,7 +435,7 @@ func newBenchmarkJob(config Config, name, secretName string, options Options, la if err := options.Validate(); err != nil { return nil, fmt.Errorf("invalid benchmark options: %w", err) } - resources, err := resourceRequirements(config.Resources) + resources, err := resourceRequirements(resolvedResources) if err != nil { return nil, err } @@ -498,54 +502,6 @@ func benchmarkEnvironment(secretName string, options Options) []corev1.EnvVar { return env } -func resourceRequirements(resources Resources) (corev1.ResourceRequirements, error) { - result := corev1.ResourceRequirements{} - requests, err := parseQuantities("request", resources.CPURequest, resources.MemoryRequest) - if err != nil { - return result, err - } - limits, err := parseQuantities("limit", resources.CPULimit, resources.MemoryLimit) - if err != nil { - return result, err - } - result.Requests = requests - for _, name := range []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory} { - request, hasRequest := requests[name] - limit, hasLimit := limits[name] - if hasRequest && hasLimit && request.Cmp(limit) > 0 { - return corev1.ResourceRequirements{}, fmt.Errorf("%s request cannot exceed limit", name) - } - } - result.Limits = limits - return result, nil -} - -func parseQuantities(kind, cpu, memory string) (corev1.ResourceList, error) { - result := corev1.ResourceList{} - for name, value := range map[string]string{"cpu": cpu, "memory": memory} { - if strings.TrimSpace(value) == "" { - continue - } - quantity, err := resource.ParseQuantity(value) - if err != nil { - return nil, fmt.Errorf("invalid %s %s resource quantity: %w", kind, name, err) - } - resourceName := corev1.ResourceName(name) - if quantity.Sign() < 0 { - return nil, fmt.Errorf("%s %s resource quantity must be nonnegative", kind, name) - } - if resourceName == corev1.ResourceCPU { - // Round a copy so precision is checked without changing the value. - rounded := quantity.DeepCopy() - if !rounded.RoundUp(resource.Milli) { - return nil, fmt.Errorf("%s cpu resource quantity must use increments of 1m", kind) - } - } - result[resourceName] = quantity - } - return result, nil -} - func copyLabels(labels map[string]string) map[string]string { if len(labels) == 0 { return nil diff --git a/backend/internal/coordinator/jobs_test.go b/backend/internal/coordinator/jobs_test.go index c76222a..21ee7d1 100644 --- a/backend/internal/coordinator/jobs_test.go +++ b/backend/internal/coordinator/jobs_test.go @@ -124,6 +124,38 @@ func TestResourceCreationRejectsInvalidResourcesBeforeAPIWrite(t *testing.T) { } } +func TestResourceCreationRejectsInvalidOverridesBeforeAPIWrite(t *testing.T) { + for _, test := range []struct { + name string + overrides Resources + wantError string + }{ + {"invalid cpu", Resources{CPURequest: "invalid"}, "request cpu"}, + {"zero memory", Resources{MemoryLimit: "0"}, "limit memory resource override must be positive"}, + {"cpu request exceeds inherited limit", Resources{CPURequest: "2"}, "cpu request cannot exceed limit"}, + {"memory limit below inherited request", Resources{MemoryLimit: "64Mi"}, "memory request cannot exceed limit"}, + } { + t.Run(test.name, func(t *testing.T) { + config := validConfig() + config.Resources = Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + client := fake.NewSimpleClientset() + c, err := New(config, client) + if err != nil { + t.Fatal(err) + } + options := DefaultOptions() + options.Resources = test.overrides + _, err = c.createBenchmarkResources(context.Background(), validConnection(), options, nil) + if err == nil || !strings.Contains(err.Error(), test.wantError) { + t.Fatalf("resource error = %v; want %q", err, test.wantError) + } + if len(client.Actions()) != 0 { + t.Fatalf("invalid overrides caused Kubernetes API calls: %v", client.Actions()) + } + }) + } +} + func TestResourceCreationPreservesJobAndCleanupErrors(t *testing.T) { client := fake.NewSimpleClientset() jobErr, cleanupErr := errors.New("job failed"), errors.New("cleanup failed") @@ -610,7 +642,7 @@ func TestCreateBenchmarkJob(t *testing.T) { t.Fatalf("New() error = %v", err) } - jobRef, err := coordinator.createBenchmarkJob(context.Background(), SecretRef{Name: "plugin-bench-credentials-run-1"}, DefaultOptions(), nil) + jobRef, err := coordinator.createBenchmarkJob(context.Background(), SecretRef{Name: "plugin-bench-credentials-run-1"}, DefaultOptions(), coordinator.config.Resources, nil) if err != nil { t.Fatalf("createBenchmarkJob() error = %v", err) } @@ -644,12 +676,16 @@ func TestCreateBenchmarkResourcesCreatesSecretBeforeJob(t *testing.T) { job.UID = types.UID("job-uid-1") return false, nil, nil }) - coordinator, err := New(validConfig(), client) + config := validConfig() + config.Resources = Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + coordinator, err := New(config, client) if err != nil { t.Fatalf("New() error = %v", err) } - resources, err := coordinator.createBenchmarkResources(context.Background(), validConnection(), DefaultOptions(), nil) + options := DefaultOptions() + options.Resources = Resources{CPURequest: "500m", MemoryLimit: "1Gi"} + resources, err := coordinator.createBenchmarkResources(context.Background(), validConnection(), options, nil) if err != nil { t.Fatalf("createBenchmarkResources() error = %v", err) } @@ -669,6 +705,22 @@ func TestCreateBenchmarkResourcesCreatesSecretBeforeJob(t *testing.T) { if got := job.Spec.Template.Spec.Containers[0].Env[0].ValueFrom.SecretKeyRef.Name; got != resources.secret.Name { t.Fatalf("Job references Secret %q, want %q", got, resources.secret.Name) } + got := job.Spec.Template.Spec.Containers[0].Resources + for _, field := range []struct { + values corev1.ResourceList + name corev1.ResourceName + want string + }{ + {got.Requests, corev1.ResourceCPU, "500m"}, + {got.Limits, corev1.ResourceCPU, "1"}, + {got.Requests, corev1.ResourceMemory, "128Mi"}, + {got.Limits, corev1.ResourceMemory, "1Gi"}, + } { + value, ok := field.values[field.name] + if !ok || value.Cmp(resource.MustParse(field.want)) != 0 { + t.Fatalf("Job %s resource = %s; want %s", field.name, value.String(), field.want) + } + } } func TestCreateBenchmarkResourcesDeletesSecretWhenJobCreationFails(t *testing.T) { @@ -738,7 +790,7 @@ func TestCreateBenchmarkJobRejectsCanceledContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - _, err = coordinator.createBenchmarkJob(ctx, SecretRef{Name: "credentials-run-1"}, DefaultOptions(), nil) + _, err = coordinator.createBenchmarkJob(ctx, SecretRef{Name: "credentials-run-1"}, DefaultOptions(), coordinator.config.Resources, nil) if !errors.Is(err, context.Canceled) { t.Fatalf("error = %v, want context.Canceled", err) } @@ -754,7 +806,7 @@ func TestCreateBenchmarkJobDoesNotExposeCredentialsOnAPIError(t *testing.T) { t.Fatalf("New() error = %v", err) } - _, err = coordinator.createBenchmarkJob(context.Background(), SecretRef{Name: "credentials-run-1"}, DefaultOptions(), nil) + _, err = coordinator.createBenchmarkJob(context.Background(), SecretRef{Name: "credentials-run-1"}, DefaultOptions(), coordinator.config.Resources, nil) if err == nil || !strings.Contains(err.Error(), "Kubernetes API unavailable") { t.Fatalf("error = %v, want API error", err) } @@ -771,7 +823,7 @@ func TestNewBenchmarkJobBuildsRunnerManifest(t *testing.T) { config := validConfig() config.Resources = Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "64Mi", MemoryLimit: "128Mi"} - job, err := newBenchmarkJob(config, "pgbench-run-1", "pgbench-secret-1", options, labels) + job, err := newBenchmarkJob(config, "pgbench-run-1", "pgbench-secret-1", options, config.Resources, labels) if err != nil { t.Fatalf("newBenchmarkJob() error = %v", err) } @@ -882,7 +934,72 @@ func TestNewBenchmarkJobRejectsInvalidResourceQuantity(t *testing.T) { config := validConfig() config.Resources.CPURequest = "not-a-quantity" - if _, err := newBenchmarkJob(config, "pgbench-run-1", "pgbench-secret-1", DefaultOptions(), nil); err == nil { + if _, err := newBenchmarkJob(config, "pgbench-run-1", "pgbench-secret-1", DefaultOptions(), config.Resources, nil); err == nil { t.Fatal("newBenchmarkJob() error = nil, want invalid resource quantity error") } } + +func TestNewBenchmarkJobUsesIndependentRunResources(t *testing.T) { + config := validConfig() + config.Resources = Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + firstOptions := DefaultOptions() + firstOptions.Resources = Resources{CPURequest: "500m", MemoryLimit: "1Gi"} + secondOptions := DefaultOptions() + secondOptions.Resources = Resources{CPULimit: "2", MemoryRequest: "256Mi"} + + for _, test := range []struct { + name string + options Options + want Resources + }{ + {"first run", firstOptions, Resources{CPURequest: "500m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "1Gi"}}, + {"second run", secondOptions, Resources{CPURequest: "100m", CPULimit: "2", MemoryRequest: "256Mi", MemoryLimit: "512Mi"}}, + {"legacy caller", DefaultOptions(), config.Resources}, + } { + t.Run(test.name, func(t *testing.T) { + resolved, err := ResolveResources(config.Resources, test.options.Resources) + if err != nil { + t.Fatal(err) + } + job, err := newBenchmarkJob(config, "pgbench-run", "pgbench-secret", test.options, resolved, nil) + if err != nil { + t.Fatal(err) + } + got := job.Spec.Template.Spec.Containers[0].Resources + for _, field := range []struct { + values corev1.ResourceList + name corev1.ResourceName + want string + }{ + {got.Requests, corev1.ResourceCPU, test.want.CPURequest}, + {got.Limits, corev1.ResourceCPU, test.want.CPULimit}, + {got.Requests, corev1.ResourceMemory, test.want.MemoryRequest}, + {got.Limits, corev1.ResourceMemory, test.want.MemoryLimit}, + } { + value, ok := field.values[field.name] + if !ok || value.Cmp(resource.MustParse(field.want)) != 0 { + t.Fatalf("%s resource = %s; want %s", field.name, value.String(), field.want) + } + } + }) + } + if config.Resources != (Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"}) { + t.Fatalf("run overrides mutated deployment defaults: %+v", config.Resources) + } +} + +func TestNewBenchmarkJobPreservesInheritedZero(t *testing.T) { + config := validConfig() + config.Resources = Resources{CPURequest: "0", MemoryLimit: "0"} + job, err := newBenchmarkJob(config, "pgbench-run", "pgbench-secret", DefaultOptions(), config.Resources, nil) + if err != nil { + t.Fatal(err) + } + resources := job.Spec.Template.Spec.Containers[0].Resources + if cpu := resources.Requests.Cpu(); cpu == nil || cpu.Sign() != 0 { + t.Fatalf("inherited CPU request = %v; want zero", cpu) + } + if memory := resources.Limits.Memory(); memory == nil || memory.Sign() != 0 { + t.Fatalf("inherited memory limit = %v; want zero", memory) + } +} diff --git a/backend/internal/coordinator/resources.go b/backend/internal/coordinator/resources.go new file mode 100644 index 0000000..8a1ca24 --- /dev/null +++ b/backend/internal/coordinator/resources.go @@ -0,0 +1,112 @@ +package coordinator + +import ( + "fmt" + "strings" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" +) + +// Resources uses Kubernetes quantity strings; empty fields remain unspecified. +type Resources struct { + CPURequest string `json:"cpuRequest,omitempty"` + MemoryRequest string `json:"memoryRequest,omitempty"` + CPULimit string `json:"cpuLimit,omitempty"` + MemoryLimit string `json:"memoryLimit,omitempty"` +} + +// ResolveResources merges per-run overrides with deployment defaults by value. +// Blank overrides inherit defaults; explicit overrides must be positive. +// Inherited zero and unspecified quantities retain deployment compatibility. +func ResolveResources(defaults, overrides Resources) (Resources, error) { + provided, err := resourceRequirements(overrides) + if err != nil { + return Resources{}, fmt.Errorf("invalid resource overrides: %w", err) + } + for _, group := range []struct { + kind string + quantities corev1.ResourceList + }{ + {"request", provided.Requests}, + {"limit", provided.Limits}, + } { + for _, name := range []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory} { + if quantity, exists := group.quantities[name]; exists && quantity.Sign() == 0 { + return Resources{}, fmt.Errorf("%s %s resource override must be positive", group.kind, name) + } + } + } + + resolved := Resources{ + CPURequest: inheritResource(defaults.CPURequest, overrides.CPURequest), + MemoryRequest: inheritResource(defaults.MemoryRequest, overrides.MemoryRequest), + CPULimit: inheritResource(defaults.CPULimit, overrides.CPULimit), + MemoryLimit: inheritResource(defaults.MemoryLimit, overrides.MemoryLimit), + } + if _, err := resourceRequirements(resolved); err != nil { + return Resources{}, fmt.Errorf("invalid resolved resources: %w", err) + } + return resolved, nil +} + +func inheritResource(defaultValue, override string) string { + if value := strings.TrimSpace(override); value != "" { + return value + } + return strings.TrimSpace(defaultValue) +} + +func resourceRequirements(resources Resources) (corev1.ResourceRequirements, error) { + result := corev1.ResourceRequirements{} + requests, err := parseQuantities("request", resources.CPURequest, resources.MemoryRequest) + if err != nil { + return result, err + } + limits, err := parseQuantities("limit", resources.CPULimit, resources.MemoryLimit) + if err != nil { + return result, err + } + result.Requests = requests + for _, name := range []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory} { + request, hasRequest := requests[name] + limit, hasLimit := limits[name] + if hasRequest && hasLimit && request.Cmp(limit) > 0 { + return corev1.ResourceRequirements{}, fmt.Errorf("%s request cannot exceed limit", name) + } + } + result.Limits = limits + return result, nil +} + +func parseQuantities(kind, cpu, memory string) (corev1.ResourceList, error) { + result := corev1.ResourceList{} + for _, field := range []struct { + name corev1.ResourceName + value string + }{ + {corev1.ResourceCPU, cpu}, + {corev1.ResourceMemory, memory}, + } { + value := strings.TrimSpace(field.value) + if value == "" { + continue + } + quantity, err := resource.ParseQuantity(value) + if err != nil { + return nil, fmt.Errorf("invalid %s %s resource quantity: %w", kind, field.name, err) + } + if quantity.Sign() < 0 { + return nil, fmt.Errorf("%s %s resource quantity must be nonnegative", kind, field.name) + } + if field.name == corev1.ResourceCPU { + // Round a copy so precision is checked without changing the value. + rounded := quantity.DeepCopy() + if !rounded.RoundUp(resource.Milli) { + return nil, fmt.Errorf("%s cpu resource quantity must use increments of 1m", kind) + } + } + result[field.name] = quantity + } + return result, nil +} diff --git a/backend/internal/coordinator/resources_test.go b/backend/internal/coordinator/resources_test.go new file mode 100644 index 0000000..08fb29e --- /dev/null +++ b/backend/internal/coordinator/resources_test.go @@ -0,0 +1,100 @@ +package coordinator + +import ( + "strings" + "testing" +) + +func TestResolveResourcesInheritance(t *testing.T) { + defaults := Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + for _, test := range []struct { + name string + defaults Resources + overrides Resources + want Resources + }{ + {"no overrides", defaults, Resources{}, defaults}, + {"cpu request", defaults, Resources{CPURequest: "250m"}, Resources{CPURequest: "250m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"}}, + {"cpu limit", defaults, Resources{CPULimit: "2"}, Resources{CPURequest: "100m", CPULimit: "2", MemoryRequest: "128Mi", MemoryLimit: "512Mi"}}, + {"memory request", defaults, Resources{MemoryRequest: "256Mi"}, Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "256Mi", MemoryLimit: "512Mi"}}, + {"memory limit", defaults, Resources{MemoryLimit: "1Gi"}, Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "1Gi"}}, + {"whitespace inherits", defaults, Resources{CPURequest: " \t", CPULimit: "\n", MemoryRequest: " ", MemoryLimit: "\t"}, defaults}, + {"trim defaults and overrides", Resources{CPURequest: " 100m ", CPULimit: " 1 ", MemoryRequest: " 128Mi ", MemoryLimit: " 512Mi "}, Resources{CPURequest: " 250m ", MemoryLimit: " 1Gi "}, Resources{CPURequest: "250m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "1Gi"}}, + {"unspecified defaults", Resources{}, Resources{}, Resources{}}, + {"partial unspecified defaults", Resources{}, Resources{CPURequest: "100m"}, Resources{CPURequest: "100m"}}, + {"inherited zero remains valid", Resources{CPURequest: "0", MemoryLimit: "0"}, Resources{}, Resources{CPURequest: "0", MemoryLimit: "0"}}, + {"equal across units", defaults, Resources{CPURequest: "1000m", MemoryRequest: "1024Mi", MemoryLimit: "1Gi"}, Resources{CPURequest: "1000m", CPULimit: "1", MemoryRequest: "1024Mi", MemoryLimit: "1Gi"}}, + {"smallest cpu increment", Resources{}, Resources{CPURequest: "0.001"}, Resources{CPURequest: "0.001"}}, + } { + t.Run(test.name, func(t *testing.T) { + beforeDefaults, beforeOverrides := test.defaults, test.overrides + got, err := ResolveResources(test.defaults, test.overrides) + if err != nil || got != test.want { + t.Fatalf("ResolveResources() = %+v, %v; want %+v", got, err, test.want) + } + if test.defaults != beforeDefaults || test.overrides != beforeOverrides { + t.Fatal("resolution mutated inputs") + } + }) + } +} + +func TestResolveResourcesRejectsInvalidValues(t *testing.T) { + defaults := Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + for _, test := range []struct { + name string + defaults Resources + overrides Resources + wantError string + }{ + {"malformed cpu request", defaults, Resources{CPURequest: "invalid"}, "request cpu"}, + {"malformed cpu limit", defaults, Resources{CPULimit: "invalid"}, "limit cpu"}, + {"malformed memory request", defaults, Resources{MemoryRequest: "invalid"}, "request memory"}, + {"malformed memory limit", defaults, Resources{MemoryLimit: "invalid"}, "limit memory"}, + {"negative cpu request", defaults, Resources{CPURequest: "-1m"}, "request cpu"}, + {"negative cpu limit", defaults, Resources{CPULimit: "-1"}, "limit cpu"}, + {"negative memory request", defaults, Resources{MemoryRequest: "-1Mi"}, "request memory"}, + {"negative memory limit", defaults, Resources{MemoryLimit: "-1Gi"}, "limit memory"}, + {"zero cpu request", defaults, Resources{CPURequest: "0m"}, "request cpu resource override must be positive"}, + {"zero cpu limit", defaults, Resources{CPULimit: "0"}, "limit cpu resource override must be positive"}, + {"zero memory request", defaults, Resources{MemoryRequest: "0Mi"}, "request memory resource override must be positive"}, + {"zero memory limit", defaults, Resources{MemoryLimit: "0Gi"}, "limit memory resource override must be positive"}, + {"sub milli cpu request", defaults, Resources{CPURequest: "0.1m"}, "increments of 1m"}, + {"sub milli cpu limit", defaults, Resources{CPULimit: "1.0001"}, "increments of 1m"}, + {"cpu request exceeds inherited limit", defaults, Resources{CPURequest: "1001m"}, "cpu request cannot exceed limit"}, + {"cpu limit below inherited request", defaults, Resources{CPULimit: "99m"}, "cpu request cannot exceed limit"}, + {"memory request exceeds inherited limit", defaults, Resources{MemoryRequest: "1Gi"}, "memory request cannot exceed limit"}, + {"memory limit below inherited request", defaults, Resources{MemoryLimit: "64Mi"}, "memory request cannot exceed limit"}, + {"explicit cpu pair invalid", defaults, Resources{CPURequest: "2", CPULimit: "1000m"}, "cpu request cannot exceed limit"}, + {"explicit memory pair invalid", defaults, Resources{MemoryRequest: "1025Mi", MemoryLimit: "1Gi"}, "memory request cannot exceed limit"}, + {"invalid inherited quantity", Resources{MemoryLimit: "invalid"}, Resources{}, "limit memory"}, + {"invalid inherited pair", Resources{CPURequest: "2", CPULimit: "1"}, Resources{}, "cpu request cannot exceed limit"}, + {"negative inherited value", Resources{MemoryRequest: "-1Mi"}, Resources{}, "request memory"}, + } { + t.Run(test.name, func(t *testing.T) { + got, err := ResolveResources(test.defaults, test.overrides) + if err == nil || !strings.Contains(err.Error(), test.wantError) { + t.Fatalf("ResolveResources() error = %v; want %q", err, test.wantError) + } + if got != (Resources{}) { + t.Fatalf("failed resolution returned resources: %+v", got) + } + }) + } +} + +func TestResolveResourcesKeepsRunsIndependent(t *testing.T) { + defaults := Resources{CPURequest: "100m", CPULimit: "1", MemoryRequest: "128Mi", MemoryLimit: "512Mi"} + first, err := ResolveResources(defaults, Resources{MemoryLimit: "1Gi"}) + if err != nil { + t.Fatal(err) + } + second, err := ResolveResources(defaults, Resources{CPURequest: "500m"}) + if err != nil { + t.Fatal(err) + } + first.CPURequest = "2" + if defaults.CPURequest != "100m" || defaults.MemoryLimit != "512Mi" || second.CPURequest != "500m" || second.MemoryLimit != "512Mi" { + t.Fatalf("per-run resources leaked into defaults or another run: defaults=%+v second=%+v", defaults, second) + } +} diff --git a/backend/internal/coordinator/runs.go b/backend/internal/coordinator/runs.go index c30385c..54ad4ad 100644 --- a/backend/internal/coordinator/runs.go +++ b/backend/internal/coordinator/runs.go @@ -47,6 +47,8 @@ type Options struct { Threads int Scale int Initialize bool + // Resources contains optional per-run overrides of the deployment defaults. + Resources Resources } func (o Options) Validate() error { diff --git a/backend/internal/coordinator/runs_test.go b/backend/internal/coordinator/runs_test.go index 0ece169..06fdd2e 100644 --- a/backend/internal/coordinator/runs_test.go +++ b/backend/internal/coordinator/runs_test.go @@ -40,6 +40,25 @@ type diagnosticClient struct { check func(context.Context) } +func TestRunRejectsInvalidResourceOverridesBeforeKubernetesWrites(t *testing.T) { + config := validConfig() + config.Resources = Resources{CPURequest: "100m", CPULimit: "1"} + client := fake.NewSimpleClientset() + c, err := New(config, client) + if err != nil { + t.Fatal(err) + } + options := DefaultOptions() + options.Resources = Resources{CPURequest: "2"} + _, err = c.Run(context.Background(), validConnection(), options) + if err == nil || !strings.Contains(err.Error(), "cpu request cannot exceed limit") { + t.Fatalf("Run() error = %v; want resource validation error", err) + } + if len(client.Actions()) != 0 { + t.Fatalf("invalid resources caused Kubernetes API calls: %v", client.Actions()) + } +} + func (c diagnosticClient) CoreV1() typedcorev1.CoreV1Interface { return diagnosticCore{CoreV1Interface: c.Clientset.CoreV1(), check: c.check} } diff --git a/backend/main.go b/backend/main.go index a370166..d33ba8c 100644 --- a/backend/main.go +++ b/backend/main.go @@ -112,7 +112,7 @@ func run() error { lifecycleCtx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - benchmarkAPI := api.NewAPI(everest.GetCredentials, everest.CheckInstanceAccess, benchmarkCoordinator, api.NewRunStore(), lifecycleCtx) + benchmarkAPI := api.NewAPI(everest.GetCredentials, everest.CheckInstanceAccess, benchmarkCoordinator, api.NewRunStore(), lifecycleCtx, config.Resources) mux := http.NewServeMux() mux.HandleFunc("GET /main.js", handleBundle) diff --git a/src/BenchmarkForm.tsx b/src/BenchmarkForm.tsx index d8302f0..259c1c5 100644 --- a/src/BenchmarkForm.tsx +++ b/src/BenchmarkForm.tsx @@ -1,6 +1,9 @@ import { useState } from 'react'; import type { FormEvent } from 'react'; import { + Accordion, + AccordionDetails, + AccordionSummary, Box, Button, Card, @@ -18,7 +21,8 @@ import { initialBenchmarkFormValues, validateBenchmarkForm } from './benchmarkFo import type { BenchmarkFormField, BenchmarkFormValues } from './benchmarkFormValidation'; import type { BenchmarkTarget } from './benchmarkTarget'; -type NumericField = Exclude; +type NumericField = 'durationSeconds' | 'clients' | 'threads' | 'scale'; +type ResourceField = 'cpuRequest' | 'cpuLimit' | 'memoryRequest' | 'memoryLimit'; const NUMERIC_FIELDS: ReadonlyArray<{ field: NumericField; label: string; hint: string }> = [ { field: 'durationSeconds', label: 'Duration (seconds)', hint: 'How long the benchmark runs.' }, @@ -27,6 +31,13 @@ const NUMERIC_FIELDS: ReadonlyArray<{ field: NumericField; label: string; hint: { field: 'scale', label: 'Scale', hint: 'Dataset size on initialization (×100,000 rows).' }, ]; +const RESOURCE_FIELDS: ReadonlyArray<{ field: ResourceField; label: string; hint: string }> = [ + { field: 'cpuRequest', label: 'CPU request (millicores)', hint: '1000m = 1 CPU core.' }, + { field: 'cpuLimit', label: 'CPU limit (millicores)', hint: '1000m = 1 CPU core.' }, + { field: 'memoryRequest', label: 'Memory request (MiB)', hint: '1024 MiB = 1 GiB.' }, + { field: 'memoryLimit', label: 'Memory limit (MiB)', hint: '1024 MiB = 1 GiB.' }, +]; + interface BenchmarkFormProps { target: BenchmarkTarget; isSubmitting: boolean; @@ -36,8 +47,10 @@ interface BenchmarkFormProps { export function BenchmarkForm({ target, isSubmitting, onSubmit }: BenchmarkFormProps) { const [values, setValues] = useState({ ...initialBenchmarkFormValues }); const [touched, setTouched] = useState>>({}); + const [resourcesExpanded, setResourcesExpanded] = useState(false); const errors = validateBenchmarkForm(values); const canSubmit = Object.keys(errors).length === 0 && !isSubmitting; + const resourceError = RESOURCE_FIELDS.some(({ field }) => Boolean(errors[field])); const updateValue = (key: K, value: BenchmarkFormValues[K]) => { setValues(current => ({ ...current, [key]: value })); @@ -49,7 +62,8 @@ export function BenchmarkForm({ target, isSubmitting, onSubmit }: BenchmarkFormP const submit = (event: FormEvent) => { event.preventDefault(); - setTouched({ database: true, durationSeconds: true, clients: true, threads: true, scale: true }); + setTouched({ database: true, durationSeconds: true, clients: true, threads: true, scale: true, + cpuRequest: true, cpuLimit: true, memoryRequest: true, memoryLimit: true }); if (canSubmit) void onSubmit(values); }; @@ -121,6 +135,46 @@ export function BenchmarkForm({ target, isSubmitting, onSubmit }: BenchmarkFormP : 'Reuses the existing pgbench tables. The run fails if they don\'t exist yet.'} + + setResourcesExpanded(expanded)} + disableGutters + > + + + + + CPU and memory for the benchmark workload generator. Leave a field empty to use its deployment default. + + + {RESOURCE_FIELDS.map(({ field, label, hint }) => ( + updateValue(field, event.target.value)} + onBlur={() => markTouched(field)} + placeholder="Use deployment default" + slotProps={{ htmlInput: { min: 1, step: 1 } }} + /> + ))} + + + + diff --git a/src/BenchmarkRunResult.tsx b/src/BenchmarkRunResult.tsx index 76705e2..07c8848 100644 --- a/src/BenchmarkRunResult.tsx +++ b/src/BenchmarkRunResult.tsx @@ -86,6 +86,33 @@ export function BenchmarkRunResult({ id, run, feedback, onRetry, onDismiss }: Be )} + {run?.resources !== undefined && ( + + + Configured runner resources + + + {([ + ['CPU request', run.resources.cpuRequest], + ['CPU limit', run.resources.cpuLimit], + ['Memory request', run.resources.memoryRequest], + ['Memory limit', run.resources.memoryLimit], + ] as const).map(([label, value]) => ( + + {label} + {value || 'Not specified'} + + ))} + + + Configured limits and requests, not measured resource usage. + + + )} + {feedback && ( run[key] === undefined || typeof run[key] === 'string') && (run.status !== 'failed' || (typeof run.error === 'string' && run.error.trim() !== '')); } +function isBenchmarkRunResources(value: unknown): value is BenchmarkRunResources { + if (typeof value !== 'object' || value === null || Array.isArray(value)) return false; + const resources = value as Record; + return ['cpuRequest', 'cpuLimit', 'memoryRequest', 'memoryLimit'] + .every(key => resources[key] === undefined || typeof resources[key] === 'string'); +} + export class BenchmarkApiError extends Error { constructor( message: string, diff --git a/src/benchmarkFormValidation.ts b/src/benchmarkFormValidation.ts index 10d2bbd..f84501d 100644 --- a/src/benchmarkFormValidation.ts +++ b/src/benchmarkFormValidation.ts @@ -7,6 +7,10 @@ export type BenchmarkFormValues = { clients: string; threads: string; scale: string; + cpuRequest: string; + cpuLimit: string; + memoryRequest: string; + memoryLimit: string; initialize: boolean; }; @@ -16,6 +20,10 @@ export const initialBenchmarkFormValues: BenchmarkFormValues = { clients: '1', threads: '1', scale: '1', + cpuRequest: '', + cpuLimit: '', + memoryRequest: '', + memoryLimit: '', initialize: true, }; @@ -69,6 +77,28 @@ export function validateBenchmarkForm(values: BenchmarkFormValues): BenchmarkFor errors.threads = 'Threads cannot exceed clients.'; } + for (const [field, label] of [ + ['cpuRequest', 'CPU request'], + ['cpuLimit', 'CPU limit'], + ['memoryRequest', 'Memory request'], + ['memoryLimit', 'Memory limit'], + ] as const) { + const value = values[field].trim(); + if (value === '') continue; + if (!/^\d+$/.test(value) || !Number.isSafeInteger(Number(value)) || Number(value) < 1) { + errors[field] = `${label} must be a positive whole number.`; + } + } + + if (!errors.cpuRequest && !errors.cpuLimit && values.cpuRequest.trim() && values.cpuLimit.trim() && + Number(values.cpuRequest) > Number(values.cpuLimit)) { + errors.cpuLimit = 'CPU limit must be at least the CPU request.'; + } + if (!errors.memoryRequest && !errors.memoryLimit && values.memoryRequest.trim() && values.memoryLimit.trim() && + Number(values.memoryRequest) > Number(values.memoryLimit)) { + errors.memoryLimit = 'Memory limit must be at least the memory request.'; + } + return errors; } @@ -81,6 +111,12 @@ export function toCreateBenchmarkRunRequest( } const database = values.database.trim(); + const resources = { + ...(values.cpuRequest.trim() ? { cpuRequest: `${Number(values.cpuRequest)}m` } : {}), + ...(values.cpuLimit.trim() ? { cpuLimit: `${Number(values.cpuLimit)}m` } : {}), + ...(values.memoryRequest.trim() ? { memoryRequest: `${Number(values.memoryRequest)}Mi` } : {}), + ...(values.memoryLimit.trim() ? { memoryLimit: `${Number(values.memoryLimit)}Mi` } : {}), + }; return { target, ...(database ? { database } : {}), @@ -89,5 +125,6 @@ export function toCreateBenchmarkRunRequest( threads: Number(values.threads), scale: values.initialize ? Number(values.scale) : 1, initialize: values.initialize, + ...(Object.keys(resources).length > 0 ? { resources } : {}), }; } diff --git a/tests/benchmarkApi.test.mjs b/tests/benchmarkApi.test.mjs index ed5f63a..fe347f5 100644 --- a/tests/benchmarkApi.test.mjs +++ b/tests/benchmarkApi.test.mjs @@ -79,6 +79,16 @@ test('explicit rejection retains its message and is distinguishable from uncerta error.status === 429 && error.message === 'capacity full'); }); +test('submits resource overrides and surfaces early validation errors', async () => { + const withResources = { ...request, resources: { cpuRequest: '500m', memoryLimit: '1Gi' } }; + await assert.rejects(createBenchmarkRun(async (path, init) => { + assert.equal(path, '/api/runs'); + assert.deepEqual(JSON.parse(init.body), withResources); + return new Response('{"error":"cpu request cannot exceed limit"}', { status: 400 }); + }, withResources), error => error instanceof BenchmarkApiError && !error.outcomeUnknown && + error.status === 400 && error.message === 'cpu request cannot exceed limit'); +}); + test('accepted response succeeds and disconnects the caller abort listener', async () => { const controller = new AbortController(); let signal; diff --git a/tests/benchmarkForm.test.mjs b/tests/benchmarkForm.test.mjs index 91eb347..25ee77d 100644 --- a/tests/benchmarkForm.test.mjs +++ b/tests/benchmarkForm.test.mjs @@ -61,3 +61,42 @@ test('enabling initialization again requires a valid scale and uses its value', assert.equal(toCreateBenchmarkRunRequest({ ...values, initialize: true }, target), undefined); assert.equal(toCreateBenchmarkRunRequest({ ...values, initialize: true, scale: '10' }, target).scale, 10); }); + +test('resource fields are optional and omitted when empty', () => { + for (const field of ['cpuRequest', 'cpuLimit', 'memoryRequest', 'memoryLimit']) { + assert.equal(initialBenchmarkFormValues[field], ''); + } + const request = toCreateBenchmarkRunRequest(validValues, target); + assert.ok(request); + assert.equal('resources' in request, false); + assert.equal('resources' in toCreateBenchmarkRunRequest({ ...validValues, cpuRequest: ' ' }, target), false); +}); + +test('partial and full resource overrides use Kubernetes units', () => { + assert.deepEqual(toCreateBenchmarkRunRequest({ + ...validValues, cpuRequest: '500', memoryLimit: '1024', + }, target).resources, { cpuRequest: '500m', memoryLimit: '1024Mi' }); + assert.deepEqual(toCreateBenchmarkRunRequest({ + ...validValues, cpuRequest: '500', cpuLimit: '1000', memoryRequest: '128', memoryLimit: '512', + }, target).resources, { + cpuRequest: '500m', cpuLimit: '1000m', memoryRequest: '128Mi', memoryLimit: '512Mi', + }); +}); + +test('resource overrides require positive, safe whole numbers', () => { + for (const field of ['cpuRequest', 'cpuLimit', 'memoryRequest', 'memoryLimit']) { + for (const invalid of ['0', '-1', '1.5', 'abc', '1e3', '9007199254740992']) { + const values = { ...validValues, [field]: invalid }; + assert.match(validateBenchmarkForm(values)[field], /positive whole number/); + assert.equal(toCreateBenchmarkRunRequest(values, target), undefined); + } + } +}); + +test('entered request cannot exceed entered limit; inherited pairs are checked by backend', () => { + assert.match(validateBenchmarkForm({ ...validValues, cpuRequest: '1001', cpuLimit: '1000' }).cpuLimit, + /at least the CPU request/); + assert.match(validateBenchmarkForm({ ...validValues, memoryRequest: '513', memoryLimit: '512' }).memoryLimit, + /at least the memory request/); + assert.deepEqual(validateBenchmarkForm({ ...validValues, cpuRequest: '1001', memoryLimit: '64' }), {}); +}); diff --git a/tests/benchmarkRunPolling.test.mjs b/tests/benchmarkRunPolling.test.mjs index 0e1ee34..8642c62 100644 --- a/tests/benchmarkRunPolling.test.mjs +++ b/tests/benchmarkRunPolling.test.mjs @@ -63,6 +63,14 @@ test('requests never overlap and terminal success/failure stops polling', async } }); +test('polling retains each run resource snapshot', async () => { + const resources = { cpuRequest: '500m', memoryLimit: '1Gi' }; + const h = harness(async () => ({ ...snapshot('succeeded'), resources })); + h.start(); + await flush(); + assert.deepEqual(h.runs[0].resources, resources); +}); + test('retry exhaustion pauses after three retries; restarting uses the same ID', async () => { const h = harness(async () => { throw new BenchmarkStatusError('offline', undefined, true); }); h.start(); diff --git a/tests/benchmarkRunResult.test.mjs b/tests/benchmarkRunResult.test.mjs index 630a6ea..03f9984 100644 --- a/tests/benchmarkRunResult.test.mjs +++ b/tests/benchmarkRunResult.test.mjs @@ -27,6 +27,27 @@ test('shows the database the run used', () => { assert.doesNotMatch(render(base), /Database/); }); +test('shows each run resource snapshot without implying measured usage', () => { + const withoutStyles = html => html.replace(/]*>[\s\S]*?<\/style>/g, ''); + const first = withoutStyles(render({ ...base, resources: { + cpuRequest: '500m', cpuLimit: '1', memoryRequest: '128Mi', memoryLimit: '1Gi', + } })); + assert.match(first, /Configured runner resources/); + assert.match(first, /CPU request<\/dt>]*>500m<\/dd>/); + assert.match(first, /Memory limit<\/dt>]*>1Gi<\/dd>/); + assert.match(first, /not measured resource usage/); + + const second = withoutStyles(render({ ...base, resources: { cpuRequest: '250m' } })); + assert.match(second, /CPU request<\/dt>]*>250m<\/dd>/); + assert.match(second, /Memory limit<\/dt>]*>Not specified<\/dd>/); + assert.doesNotMatch(second, /500m|1Gi/); + assert.match(render({ ...base, status: 'running', completedAt: undefined, resources: { cpuRequest: '250m' } }), + /Configured runner resources/); + assert.match(render({ ...base, status: 'failed', error: 'runner failed', resources: { cpuRequest: '250m' } }), + /Configured runner resources/); + assert.doesNotMatch(render(base), /Configured runner resources/); +}); + test('empty completed output is explicit and running output is not displayed', () => { assert.match(render(base), /No output was captured/); const html = render({ ...base, status: 'running', completedAt: undefined }); diff --git a/tests/benchmarkStatus.test.mjs b/tests/benchmarkStatus.test.mjs index f2126da..e8edb53 100644 --- a/tests/benchmarkStatus.test.mjs +++ b/tests/benchmarkStatus.test.mjs @@ -20,10 +20,19 @@ test('GET encodes the ID and accepts all statuses and optional diagnostics', asy } }); +test('status accepts optional resolved resource snapshots without changing them', async () => { + const resources = { cpuRequest: '500m', cpuLimit: '1', memoryRequest: '128Mi', memoryLimit: '1Gi' }; + for (const run of [running, { ...running, resources }, { ...running, resources: {} }]) { + assert.deepEqual(await getBenchmarkRun(async () => response(run), running.id), run); + } +}); + test('invalid JSON, shapes, IDs and terminal fields are rejected', async () => { for (const body of [null, {}, { ...running, id: 'other' }, { ...running, status: 'pending' }, { ...running, createdAt: 'invalid' }, { ...running, outputTruncated: 'false' }, { ...running, output: 123 }, { ...running, completedAt: null }, + { ...running, resources: null }, { ...running, resources: [] }, + { ...running, resources: { cpuRequest: 500 } }, { ...running, resources: { memoryLimit: null } }, { ...running, status: 'succeeded' }, { ...running, status: 'failed', completedAt: running.createdAt }]) { await assert.rejects(getBenchmarkRun(async () => response(body), running.id),