Skip to content

Feat/runner resources #51

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amh1k merged 5 commits into from
Oct 14, 2026
Merged

Feat/runner resources #51

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
77b0b56
feat(coordinator): add resource override resolver
amh1k Oct 7, 2026
36d5d35
feat(backend): accept and apply per-run runner resources
amh1k Oct 8, 2026
383e57f
feat(ui): configure and display benchmark job resources
amh1k Oct 8, 2026
28f630c
Merge branch 'main' into feat/runner-resources
amh1k Oct 8, 2026
a6d49f9
refactor: share resource types and address resource UI review feedback
amh1k Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion backend/internal/api/routes.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ type API struct {
getCredentials CredentialLookup
checkInstanceAccess InstanceAccessCheck
runner BenchmarkRunner
resourceDefaults coordinator.Resources
store *RunStore
lifecycleCtx context.Context
runSlots chan struct{}
Expand All @@ -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()
}
Expand All @@ -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),
Expand Down
77 changes: 42 additions & 35 deletions backend/internal/api/runs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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,
Expand Down
138 changes: 129 additions & 9 deletions backend/internal/api/runs_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"log"
"net/http"
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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},
Expand All @@ -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) {
Expand All @@ -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")
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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)
}
Expand All @@ -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
}
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading