From 70311672495ff027878c225be85a2c817dcaf16f Mon Sep 17 00:00:00 2001 From: amh1k Date: Sat, 19 Sep 2026 18:22:46 +0500 Subject: [PATCH 1/3] feat(coordinator): define benchmark run contract Signed-off-by: amh1k --- backend/internal/coordinator/coordinator.go | 63 +++++++++++ backend/internal/coordinator/runs.go | 110 ++++++++++++++++++++ 2 files changed, 173 insertions(+) create mode 100644 backend/internal/coordinator/coordinator.go create mode 100644 backend/internal/coordinator/runs.go diff --git a/backend/internal/coordinator/coordinator.go b/backend/internal/coordinator/coordinator.go new file mode 100644 index 0000000..2fb7f35 --- /dev/null +++ b/backend/internal/coordinator/coordinator.go @@ -0,0 +1,63 @@ +package coordinator + +import ( + "errors" + "strings" + "time" +) + +// Config contains deployment-level settings shared by benchmark runs. +type Config struct { + WorkloadNamespace string + RunnerImage string + ImagePullPolicy string + RunnerServiceAccount string + Resources Resources + ExecutionTimeout time.Duration +} + +func (c Config) Validate() error { + if strings.TrimSpace(c.WorkloadNamespace) == "" { + return errors.New("workload namespace is required") + } + if strings.TrimSpace(c.RunnerImage) == "" { + return errors.New("runner image is required") + } + switch c.ImagePullPolicy { + case "Always", "IfNotPresent", "Never": + default: + return errors.New("image pull policy must be Always, IfNotPresent, or Never") + } + if strings.TrimSpace(c.RunnerServiceAccount) == "" { + return errors.New("runner service account is required") + } + if c.ExecutionTimeout < time.Second || c.ExecutionTimeout%time.Second != 0 { + return errors.New("execution timeout must be a positive whole number of seconds") + } + return nil +} + +// Resources uses Kubernetes quantity strings, such as "100m" and "128Mi". +// Empty fields leave that request or limit unspecified. Quantity parsing and +// request/limit validation belong to Job construction when Kubernetes types +// are introduced. +type Resources struct { + CPURequest string + MemoryRequest string + CPULimit string + MemoryLimit string +} + +// Coordinator holds shared configuration, not per-run credentials or options. +// Kubernetes client support is added in the next implementation step. +type Coordinator struct { + config Config +} + +// New validates the shared configuration before constructing a coordinator. +func New(config Config) (*Coordinator, error) { + if err := config.Validate(); err != nil { + return nil, err + } + return &Coordinator{config: config}, nil +} diff --git a/backend/internal/coordinator/runs.go b/backend/internal/coordinator/runs.go new file mode 100644 index 0000000..ee85ff2 --- /dev/null +++ b/backend/internal/coordinator/runs.go @@ -0,0 +1,110 @@ +package coordinator + +import ( + "context" + "errors" + "fmt" + "strings" +) + +var ErrRunNotImplemented = errors.New("benchmark execution is not implemented") + +// Connection contains resolved database credentials. Do not log this value. +// Database must explicitly identify the database used for benchmarking. +type Connection struct { + Host string + Port int + Database string + Username string + Password string +} + +func (c Connection) Validate() error { + if strings.TrimSpace(c.Host) == "" { + return errors.New("database host is required") + } + if c.Port < 1 || c.Port > 65535 { + return errors.New("database port must be between 1 and 65535") + } + if strings.TrimSpace(c.Database) == "" { + return errors.New("database name is required") + } + if strings.TrimSpace(c.Username) == "" { + return errors.New("database username is required") + } + if c.Password == "" { + return errors.New("database password is required") + } + return nil +} + +// Options contains per-run settings. Duration is measured in seconds. +// Initialize must be explicitly enabled: it drops and recreates pgbench tables. +type Options struct { + Duration int + Clients int + Threads int + Scale int + Initialize bool +} + +func (o Options) Validate() error { + // Match the runner's supported positive 31-bit integer range. + const maximum = 1<<31 - 1 + if o.Duration < 1 || o.Duration > maximum { + return errors.New("benchmark duration must be between 1 and 2147483647 seconds") + } + if o.Clients < 1 || o.Clients > maximum { + return errors.New("benchmark clients must be between 1 and 2147483647") + } + if o.Threads < 1 || o.Threads > maximum { + return errors.New("benchmark threads must be between 1 and 2147483647") + } + if o.Threads > o.Clients { + return errors.New("benchmark threads cannot exceed clients") + } + if o.Scale < 1 || o.Scale > maximum { + return errors.New("benchmark scale must be between 1 and 2147483647") + } + return nil +} + +// Result contains execution output, not parsed metrics or persisted history. +// A run will return this alongside an error when execution or cleanup fails. +type Result struct { + JobName string + Output string + OutputTruncated bool +} + +// Run validates one benchmark request. Kubernetes Secret and Job creation +// will be added when the execution components are implemented. +func (c *Coordinator) Run(ctx context.Context, connection Connection, options Options) (Result, error) { + var result Result + + if c == nil { + return result, errors.New("coordinator is nil") + } + if ctx == nil { + return result, errors.New("run context is nil") + } + if err := c.config.Validate(); err != nil { + return result, fmt.Errorf("invalid coordinator configuration: %w", err) + } + if err := connection.Validate(); err != nil { + return result, fmt.Errorf("invalid database connection: %w", err) + } + if err := options.Validate(); err != nil { + return result, fmt.Errorf("invalid benchmark options: %w", err) + } + if err := ctx.Err(); err != nil { + return result, err + } + + return result, ErrRunNotImplemented +} + +// DefaultOptions matches the existing pgbench runner defaults. +func DefaultOptions() Options { + return Options{Duration: 60, Clients: 1, Threads: 1, Scale: 1} +} From 82eb58c203fd810cd19d5cca1888b05fdb2a56ee Mon Sep 17 00:00:00 2001 From: amh1k Date: Sat, 19 Sep 2026 19:46:18 +0500 Subject: [PATCH 2/3] feat(coordinator): add Kubernetes client and Job and Secret manifest builders Signed-off-by: amh1k --- backend/go.mod | 45 ++++- backend/go.sum | 154 ++++++++++++++++ backend/internal/coordinator/coordinator.go | 36 +++- .../internal/coordinator/coordinator_test.go | 57 ++++++ backend/internal/coordinator/jobs.go | 164 ++++++++++++++++++ backend/internal/coordinator/jobs_test.go | 132 ++++++++++++++ backend/internal/coordinator/secrets.go | 48 +++++ backend/internal/coordinator/secrets_test.go | 66 +++++++ 8 files changed, 696 insertions(+), 6 deletions(-) create mode 100644 backend/go.sum create mode 100644 backend/internal/coordinator/coordinator_test.go create mode 100644 backend/internal/coordinator/jobs.go create mode 100644 backend/internal/coordinator/jobs_test.go create mode 100644 backend/internal/coordinator/secrets.go create mode 100644 backend/internal/coordinator/secrets_test.go diff --git a/backend/go.mod b/backend/go.mod index 79eb3cc..08fae37 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -1,3 +1,46 @@ module github.com/openeverest/plugin-bench/backend -go 1.22 +go 1.22.0 + +require k8s.io/client-go v0.30.14 + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/emicklei/go-restful/v3 v3.11.0 // indirect + github.com/evanphx/json-patch v4.12.0+incompatible // indirect + github.com/go-logr/logr v1.4.1 // indirect + github.com/go-openapi/jsonpointer v0.19.6 // indirect + github.com/go-openapi/jsonreference v0.20.2 // indirect + github.com/go-openapi/swag v0.22.3 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang/protobuf v1.5.4 // indirect + github.com/google/gnostic-models v0.6.8 // indirect + github.com/google/gofuzz v1.2.0 // indirect + github.com/google/uuid v1.3.0 // indirect + github.com/josharian/intern v1.0.0 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/mailru/easyjson v0.7.7 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/pkg/errors v0.9.1 // indirect + golang.org/x/net v0.23.0 // indirect + golang.org/x/oauth2 v0.10.0 // indirect + golang.org/x/sys v0.18.0 // indirect + golang.org/x/term v0.18.0 // indirect + golang.org/x/text v0.14.0 // indirect + golang.org/x/time v0.3.0 // indirect + google.golang.org/appengine v1.6.7 // indirect + google.golang.org/protobuf v1.33.0 // indirect + gopkg.in/inf.v0 v0.9.1 // indirect + gopkg.in/yaml.v2 v2.4.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect + k8s.io/api v0.30.14 // indirect + k8s.io/apimachinery v0.30.14 // indirect + k8s.io/klog/v2 v2.120.1 // indirect + k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect + k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect + sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect + sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect + sigs.k8s.io/yaml v1.3.0 // indirect +) diff --git a/backend/go.sum b/backend/go.sum new file mode 100644 index 0000000..72b76cd --- /dev/null +++ b/backend/go.sum @@ -0,0 +1,154 @@ +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g= +github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= +github.com/evanphx/json-patch v4.12.0+incompatible h1:4onqiflcdA9EOZ4RxV643DvftH5pOlLGNtQ5lPWQu84= +github.com/evanphx/json-patch v4.12.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk= +github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ= +github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-openapi/jsonpointer v0.19.6 h1:eCs3fxoIi3Wh6vtgmLTOjdhSpiqphQ+DaPn38N2ZdrE= +github.com/go-openapi/jsonpointer v0.19.6/go.mod h1:osyAmYz/mB/C3I+WsTTSgw1ONzaLJoLCyoi6/zppojs= +github.com/go-openapi/jsonreference v0.20.2 h1:3sVjiK66+uXK/6oQ8xgcRKcFgQ5KXa2KvnJRumpMGbE= +github.com/go-openapi/jsonreference v0.20.2/go.mod h1:Bl1zwGIM8/wsvqjsOQLJ/SH+En5Ap4rVB5KVcIDZG2k= +github.com/go-openapi/swag v0.22.3 h1:yMBqmnQ0gyZvEb/+KzuWZOXgllrXT4SADYbvDaXHv/g= +github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14= +github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI= +github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/gnostic-models v0.6.8 h1:yo/ABAfM5IMRsS1VnXjTBvUb61tFIHozhlYvRgGre9I= +github.com/google/gnostic-models v0.6.8/go.mod h1:5n7qKqH0f5wFt+aWF8CW6pZLLNOfYuF5OpfBSENuI8U= +github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0= +github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJYCmNdQXq6neHJOYx3V6jnqNEec= +github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE= +github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= +github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= +github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= +github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/onsi/ginkgo/v2 v2.15.0 h1:79HwNRBAZHOEwrczrgSOPy+eFTTlIGELKy5as+ClttY= +github.com/onsi/ginkgo/v2 v2.15.0/go.mod h1:HlxMHtYF57y6Dpf+mc5529KKmSq9h2FpCF+/ZkwUxKM= +github.com/onsi/gomega v1.31.0 h1:54UJxxj6cPInHS3a35wm6BK/F9nHYueZ1NVujHDrnXE= +github.com/onsi/gomega v1.31.0/go.mod h1:DW9aCi7U6Yi40wNVAvT6kzFnEVEI5n3DloYBiKiT6zk= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= +github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs= +golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg= +golang.org/x/oauth2 v0.10.0 h1:zHCpF2Khkwy4mMB4bv0U37YtJdTGW8jI0glAApi0Kh8= +golang.org/x/oauth2 v0.10.0/go.mod h1:kTpgurOux7LqtuxjuyZa4Gj2gdezIt/jQtGnNFfypQI= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4= +golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/term v0.18.0 h1:FcHjZXDMxI8mM3nwhX9HlKop4C0YQvCVCdwYl2wOtE8= +golang.org/x/term v0.18.0/go.mod h1:ILwASektA3OnRv7amZ1xhE/KTR+u50pbXfZ03+6Nx58= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4= +golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.18.0 h1:k8NLag8AGHnn+PHbl7g43CtqZAwG60vZkLqgyZgIHgQ= +golang.org/x/tools v0.18.0/go.mod h1:GL7B4CwcLLeo59yx/9UWWuNOW1n3VZ4f5axWfML7Lcg= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/appengine v1.6.7 h1:FZR1q0exgwxzPzp/aF+VccGrSfxfPpkBqjIIEq3ru6c= +google.golang.org/appengine v1.6.7/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc= +google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI= +google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= +gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= +gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +k8s.io/api v0.30.14 h1:iPq9YNOz1vHcSuN9YTmRUt8iPpB1cYPxxjgbY25xfS4= +k8s.io/api v0.30.14/go.mod h1:IdrH4AiKc2bqDDb1FAfwcP1pPRmDdyRIqNk4K8KkEoc= +k8s.io/apimachinery v0.30.14 h1:2OvEYwWoWeb25+xzFGP/8gChu+MfRNv24BlCQdnfGzQ= +k8s.io/apimachinery v0.30.14/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc= +k8s.io/client-go v0.30.14 h1:D81QZvBtv897JU4HRsx4YoaCDnzeZSvB8eApgmbtXVA= +k8s.io/client-go v0.30.14/go.mod h1:9ytP3kKzrz3ZWavlWih4NB0mTdYA0DB1ElBHimq+JqQ= +k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw= +k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE= +k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 h1:BZqlfIlq5YbRMFko6/PM7FjZpUb45WallggurYhKGag= +k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340/go.mod h1:yD4MZYeKMBwQKVht279WycxKyM84kkAx2DPrTXaeb98= +k8s.io/utils v0.0.0-20230726121419-3b25d923346b h1:sgn3ZU783SCgtaSJjpcVVlRqd6GSnlTLKgpAAttJvpI= +k8s.io/utils v0.0.0-20230726121419-3b25d923346b/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= +sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd h1:EDPBXCAspyGV4jQlpZSudPeMmr1bNJefnuqLsRAsHZo= +sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0= +sigs.k8s.io/structured-merge-diff/v4 v4.4.1 h1:150L+0vs/8DA78h1u02ooW1/fFq/Lwr+sGiqlzvrtq4= +sigs.k8s.io/structured-merge-diff/v4 v4.4.1/go.mod h1:N8hJocpFajUSSeSJ9bOZ77VzejKZaXsTtZo4/u7Io08= +sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo= +sigs.k8s.io/yaml v1.3.0/go.mod h1:GeOyir5tyXNByN85N/dRIT9es5UQNerPYEKK56eTBm8= diff --git a/backend/internal/coordinator/coordinator.go b/backend/internal/coordinator/coordinator.go index 2fb7f35..f170a50 100644 --- a/backend/internal/coordinator/coordinator.go +++ b/backend/internal/coordinator/coordinator.go @@ -2,10 +2,16 @@ package coordinator import ( "errors" + "fmt" "strings" "time" + + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" ) +var ErrKubernetesClientRequired = errors.New("kubernetes client is required") + // Config contains deployment-level settings shared by benchmark runs. type Config struct { WorkloadNamespace string @@ -49,15 +55,35 @@ type Resources struct { } // Coordinator holds shared configuration, not per-run credentials or options. -// Kubernetes client support is added in the next implementation step. +// The Kubernetes client is injected so the coordinator remains testable. type Coordinator struct { - config Config + config Config + kubeClient kubernetes.Interface } -// New validates the shared configuration before constructing a coordinator. -func New(config Config) (*Coordinator, error) { +// New validates the shared configuration and injects the Kubernetes client. +// The interface keeps the coordinator testable with a fake client. +func New(config Config, kubeClient kubernetes.Interface) (*Coordinator, error) { if err := config.Validate(); err != nil { return nil, err } - return &Coordinator{config: config}, nil + if kubeClient == nil { + return nil, ErrKubernetesClientRequired + } + return &Coordinator{config: config, kubeClient: kubeClient}, nil +} + +// NewInCluster constructs a coordinator using the Pod's ServiceAccount +// credentials. Use New with an injected client in unit tests. +func NewInCluster(config Config) (*Coordinator, error) { + restConfig, err := rest.InClusterConfig() + if err != nil { + return nil, fmt.Errorf("load in-cluster Kubernetes config: %w", err) + } + + kubeClient, err := kubernetes.NewForConfig(restConfig) + if err != nil { + return nil, fmt.Errorf("create Kubernetes client: %w", err) + } + return New(config, kubeClient) } diff --git a/backend/internal/coordinator/coordinator_test.go b/backend/internal/coordinator/coordinator_test.go new file mode 100644 index 0000000..4d801a2 --- /dev/null +++ b/backend/internal/coordinator/coordinator_test.go @@ -0,0 +1,57 @@ +package coordinator + +import ( + "errors" + "testing" + "time" + + "k8s.io/client-go/kubernetes/fake" +) + +func validConfig() Config { + return Config{ + WorkloadNamespace: "plugin-bench-workloads", + RunnerImage: "ghcr.io/openeverest/plugin-bench-pgbench:dev", + ImagePullPolicy: "IfNotPresent", + RunnerServiceAccount: "plugin-bench-runner", + ExecutionTimeout: time.Minute, + } +} + +func TestNewInjectsKubernetesClient(t *testing.T) { + client := fake.NewSimpleClientset() + + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + if coordinator == nil { + t.Fatal("New() returned a nil coordinator") + } + if coordinator.kubeClient != client { + t.Fatal("New() did not retain the injected Kubernetes client") + } +} + +func TestNewRequiresKubernetesClient(t *testing.T) { + coordinator, err := New(validConfig(), nil) + if coordinator != nil { + t.Fatal("New() returned a coordinator without a Kubernetes client") + } + if !errors.Is(err, ErrKubernetesClientRequired) { + t.Fatalf("New() error = %v, want ErrKubernetesClientRequired", err) + } +} + +func TestNewValidatesConfigBeforeClient(t *testing.T) { + invalid := validConfig() + invalid.WorkloadNamespace = "" + + coordinator, err := New(invalid, nil) + if coordinator != nil { + t.Fatal("New() returned a coordinator for invalid configuration") + } + if err == nil || err.Error() != "workload namespace is required" { + t.Fatalf("New() error = %v, want workload namespace validation error", err) + } +} diff --git a/backend/internal/coordinator/jobs.go b/backend/internal/coordinator/jobs.go new file mode 100644 index 0000000..65cad85 --- /dev/null +++ b/backend/internal/coordinator/jobs.go @@ -0,0 +1,164 @@ +package coordinator + +import ( + "errors" + "fmt" + "strconv" + "strings" + + 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" +) + +const runnerContainerName = "pgbench-runner" + +// 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) { + if err := config.Validate(); err != nil { + return nil, fmt.Errorf("invalid coordinator configuration: %w", err) + } + if strings.TrimSpace(name) == "" { + return nil, errors.New("job name is required") + } + if strings.TrimSpace(secretName) == "" { + return nil, errors.New("credential secret name is required") + } + if err := options.Validate(); err != nil { + return nil, fmt.Errorf("invalid benchmark options: %w", err) + } + resources, err := resourceRequirements(config.Resources) + if err != nil { + return nil, err + } + + backoffLimit := int32(0) + activeDeadlineSeconds := int64(config.ExecutionTimeout.Seconds()) + return &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: config.WorkloadNamespace, + Labels: copyLabels(labels), + }, + Spec: batchv1.JobSpec{ + BackoffLimit: &backoffLimit, + ActiveDeadlineSeconds: &activeDeadlineSeconds, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: copyLabels(labels)}, + Spec: corev1.PodSpec{ + RestartPolicy: corev1.RestartPolicyNever, + ServiceAccountName: config.RunnerServiceAccount, + AutomountServiceAccountToken: boolPtr(false), + Containers: []corev1.Container{{ + Name: runnerContainerName, + Image: config.RunnerImage, + ImagePullPolicy: corev1.PullPolicy(config.ImagePullPolicy), + Resources: resources, + Env: benchmarkEnvironment(secretName, options), + }}, + }, + }, + }, + }, nil +} + +func benchmarkEnvironment(secretName string, options Options) []corev1.EnvVar { + // Kubernetes copies each matching field from the temporary Secret into an + // environment variable in the runner container. The names are identical + // because the runner reads these exact variables: PGHOST, PGPORT, + // PGDATABASE, PGUSER, and PGPASSWORD. + secretKeys := []string{ + secretKeyHost, + secretKeyPort, + secretKeyDatabase, + secretKeyUser, + secretKeyPassword, + } + env := make([]corev1.EnvVar, 0, len(secretKeys)+5) + for _, key := range secretKeys { + env = append(env, corev1.EnvVar{ + Name: key, + ValueFrom: &corev1.EnvVarSource{SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: secretName}, + Key: key, + }}, + }) + } + env = append(env, + corev1.EnvVar{Name: "BENCH_DURATION", Value: strconv.Itoa(options.Duration)}, + corev1.EnvVar{Name: "BENCH_CLIENTS", Value: strconv.Itoa(options.Clients)}, + corev1.EnvVar{Name: "BENCH_THREADS", Value: strconv.Itoa(options.Threads)}, + corev1.EnvVar{Name: "BENCH_SCALE", Value: strconv.Itoa(options.Scale)}, + corev1.EnvVar{Name: "BENCH_INITIALIZE", Value: strconv.FormatBool(options.Initialize)}, + ) + 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 + } + result := make(map[string]string, len(labels)) + for key, value := range labels { + result[key] = value + } + return result +} + +func boolPtr(value bool) *bool { + return &value +} + +func formatPort(port int) string { + return strconv.Itoa(port) +} diff --git a/backend/internal/coordinator/jobs_test.go b/backend/internal/coordinator/jobs_test.go new file mode 100644 index 0000000..c0d29c1 --- /dev/null +++ b/backend/internal/coordinator/jobs_test.go @@ -0,0 +1,132 @@ +package coordinator + +import ( + "testing" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" +) + +func TestNewBenchmarkJobBuildsRunnerManifest(t *testing.T) { + labels := map[string]string{"plugin-bench/run": "run-1"} + options := DefaultOptions() + options.Clients = 2 + options.Threads = 2 + + 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) + if err != nil { + t.Fatalf("newBenchmarkJob() error = %v", err) + } + if job.Namespace != "plugin-bench-workloads" || job.Name != "pgbench-run-1" { + t.Fatalf("job identity = %s/%s", job.Namespace, job.Name) + } + if job.Spec.Template.Spec.ServiceAccountName != "plugin-bench-runner" { + t.Fatalf("service account = %q", job.Spec.Template.Spec.ServiceAccountName) + } + if job.Spec.Template.Spec.RestartPolicy != corev1.RestartPolicyNever { + t.Fatalf("restart policy = %q", job.Spec.Template.Spec.RestartPolicy) + } + if job.Spec.BackoffLimit == nil || *job.Spec.BackoffLimit != 0 { + t.Fatal("Job must disable retries") + } + if job.Spec.ActiveDeadlineSeconds == nil || *job.Spec.ActiveDeadlineSeconds != 60 { + t.Fatal("Job must use the configured execution deadline") + } + if token := job.Spec.Template.Spec.AutomountServiceAccountToken; token == nil || *token { + t.Fatal("runner must disable ServiceAccount token mounting") + } + if len(job.Spec.Template.Spec.Containers) != 1 { + t.Fatal("Job must contain exactly one runner container") + } + container := job.Spec.Template.Spec.Containers[0] + if container.Image != validConfig().RunnerImage { + t.Fatalf("runner image = %q", container.Image) + } + if len(container.Env) != 10 { + t.Fatalf("environment variable count = %d, want 10", len(container.Env)) + } + env := make(map[string]corev1.EnvVar) + for _, variable := range container.Env { + if _, exists := env[variable.Name]; exists { + t.Fatalf("duplicate environment variable %s", variable.Name) + } + env[variable.Name] = variable + } + for _, name := range []string{"PGHOST", "PGPORT", "PGDATABASE", "PGUSER", "PGPASSWORD"} { + variable := env[name] + if variable.Value != "" || variable.ValueFrom == nil || variable.ValueFrom.SecretKeyRef == nil { + t.Fatalf("%s must use a Secret reference without a literal value", name) + } + ref := variable.ValueFrom.SecretKeyRef + if ref.Name != "pgbench-secret-1" || ref.Key != name || (ref.Optional != nil && *ref.Optional) { + t.Fatalf("%s has an incorrect or optional Secret reference", name) + } + } + for name, want := range map[string]string{ + "BENCH_DURATION": "60", "BENCH_CLIENTS": "2", "BENCH_THREADS": "2", + "BENCH_SCALE": "1", "BENCH_INITIALIZE": "false", + } { + if variable, ok := env[name]; !ok || variable.Value != want || variable.ValueFrom != nil { + t.Fatalf("incorrect benchmark environment variable %s", name) + } + } + for _, check := range []struct { + values corev1.ResourceList + name corev1.ResourceName + want string + }{ + {container.Resources.Requests, corev1.ResourceCPU, "100m"}, + {container.Resources.Limits, corev1.ResourceCPU, "1"}, + {container.Resources.Requests, corev1.ResourceMemory, "64Mi"}, + {container.Resources.Limits, corev1.ResourceMemory, "128Mi"}, + } { + got, ok := check.values[check.name] + if !ok || got.Cmp(resource.MustParse(check.want)) != 0 { + t.Fatalf("incorrect %s resource value, want %s", check.name, check.want) + } + } + if job.Labels["plugin-bench/run"] != "run-1" || job.Spec.Template.Labels["plugin-bench/run"] != "run-1" { + t.Fatal("run labels were not copied to Job and Pod") + } +} + +func TestResourceRequirementsValidation(t *testing.T) { + for _, test := range []struct { + name string + resources Resources + wantError bool + }{ + {"empty", Resources{}, false}, + {"zero", Resources{CPURequest: "0", MemoryLimit: "0"}, false}, + {"request only", Resources{CPURequest: "100m"}, false}, + {"limit only", Resources{MemoryLimit: "1Gi"}, false}, + {"equal across units", Resources{CPURequest: "1000m", CPULimit: "1", MemoryRequest: "1024Mi", MemoryLimit: "1Gi"}, false}, + {"negative cpu request", Resources{CPURequest: "-1"}, true}, + {"negative cpu limit", Resources{CPULimit: "-100m"}, true}, + {"negative memory request", Resources{MemoryRequest: "-1Mi"}, true}, + {"negative memory limit", Resources{MemoryLimit: "-1Gi"}, true}, + {"cpu request above limit", Resources{CPURequest: "2", CPULimit: "1"}, true}, + {"memory request above limit", Resources{MemoryRequest: "1Gi", MemoryLimit: "512Mi"}, true}, + {"sub milli request", Resources{CPURequest: "0.1m"}, true}, + {"sub milli limit", Resources{CPULimit: "1.0001"}, true}, + {"milli boundary", Resources{CPURequest: "0.001"}, false}, + } { + t.Run(test.name, func(t *testing.T) { + _, err := resourceRequirements(test.resources) + if (err != nil) != test.wantError { + t.Fatalf("error = %v, wantError = %v", err, test.wantError) + } + }) + } +} + +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 { + t.Fatal("newBenchmarkJob() error = nil, want invalid resource quantity error") + } +} diff --git a/backend/internal/coordinator/secrets.go b/backend/internal/coordinator/secrets.go new file mode 100644 index 0000000..d0d5cec --- /dev/null +++ b/backend/internal/coordinator/secrets.go @@ -0,0 +1,48 @@ +package coordinator + +import ( + "errors" + "strings" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +const ( + secretKeyHost = "PGHOST" + secretKeyPort = "PGPORT" + secretKeyDatabase = "PGDATABASE" + secretKeyUser = "PGUSER" + secretKeyPassword = "PGPASSWORD" +) + +// newCredentialSecret builds the per-run credential Secret. It only creates +// an in-memory Kubernetes object; the caller is responsible for API calls and +// cleanup. +func newCredentialSecret(namespace, name string, connection Connection, labels map[string]string) (*corev1.Secret, error) { + if strings.TrimSpace(namespace) == "" { + return nil, errors.New("secret namespace is required") + } + if strings.TrimSpace(name) == "" { + return nil, errors.New("secret name is required") + } + if err := connection.Validate(); err != nil { + return nil, err + } + + return &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: copyLabels(labels), + }, + Type: corev1.SecretTypeOpaque, + Data: map[string][]byte{ + secretKeyHost: []byte(connection.Host), + secretKeyPort: []byte(formatPort(connection.Port)), + secretKeyDatabase: []byte(connection.Database), + secretKeyUser: []byte(connection.Username), + secretKeyPassword: []byte(connection.Password), + }, + }, nil +} diff --git a/backend/internal/coordinator/secrets_test.go b/backend/internal/coordinator/secrets_test.go new file mode 100644 index 0000000..0482f57 --- /dev/null +++ b/backend/internal/coordinator/secrets_test.go @@ -0,0 +1,66 @@ +package coordinator + +import ( + "strings" + "testing" +) + +func TestNewCredentialSecretRejectsInvalidInput(t *testing.T) { + for _, field := range []string{"namespace", "name", "host", "port", "database", "username", "password"} { + t.Run(field, func(t *testing.T) { + namespace, name := "workloads", "credentials" + connection := Connection{Host: "postgres", Port: 5432, Database: "bench", Username: "postgres", Password: "sensitive-test-password"} + switch field { + case "namespace": + namespace = " " + case "name": + name = " " + case "host": + connection.Host = "" + case "port": + connection.Port = 65536 + case "database": + connection.Database = "" + case "username": + connection.Username = "" + case "password": + connection.Password = "" + } + secret, err := newCredentialSecret(namespace, name, connection, nil) + if err == nil || secret != nil { + t.Fatal("invalid input must return an error and no Secret") + } + if strings.Contains(err.Error(), "sensitive-test-password") { + t.Fatal("validation error exposed credentials") + } + }) + } +} + +func TestNewCredentialSecretBuildsConnectionData(t *testing.T) { + connection := Connection{ + Host: "postgres.example.com", + Port: 5432, + Database: "pgbench", + Username: "postgres", + Password: "secret", + } + + secret, err := newCredentialSecret("plugin-bench-workloads", "pgbench-secret-1", connection, nil) + if err != nil { + t.Fatalf("newCredentialSecret() error = %v", err) + } + if secret.Namespace != "plugin-bench-workloads" || secret.Name != "pgbench-secret-1" { + t.Fatalf("secret identity = %s/%s", secret.Namespace, secret.Name) + } + if secret.Type != "Opaque" { + t.Fatalf("secret type = %q", secret.Type) + } + if string(secret.Data[secretKeyHost]) != connection.Host || + string(secret.Data[secretKeyPort]) != "5432" || + string(secret.Data[secretKeyDatabase]) != connection.Database || + string(secret.Data[secretKeyUser]) != connection.Username || + string(secret.Data[secretKeyPassword]) != connection.Password { + t.Fatal("secret does not contain the expected connection data") + } +} From e1f704b17fbbe2d4dbbd503b949186be2a15b919 Mon Sep 17 00:00:00 2001 From: amh1k Date: Sun, 20 Sep 2026 11:03:38 +0500 Subject: [PATCH 3/3] feat(coordinator): manage benchmark Job and credential Secret creation Signed-off-by: amh1k --- backend/go.mod | 8 +- backend/internal/coordinator/jobs.go | 119 ++++++- backend/internal/coordinator/jobs_test.go | 307 +++++++++++++++++++ backend/internal/coordinator/secrets.go | 105 ++++++- backend/internal/coordinator/secrets_test.go | 146 +++++++++ 5 files changed, 676 insertions(+), 9 deletions(-) diff --git a/backend/go.mod b/backend/go.mod index 08fae37..166775e 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -2,7 +2,11 @@ module github.com/openeverest/plugin-bench/backend go 1.22.0 -require k8s.io/client-go v0.30.14 +require ( + k8s.io/api v0.30.14 + k8s.io/apimachinery v0.30.14 + k8s.io/client-go v0.30.14 +) require ( github.com/davecgh/go-spew v1.1.1 // indirect @@ -35,8 +39,6 @@ require ( gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect - k8s.io/api v0.30.14 // indirect - k8s.io/apimachinery v0.30.14 // indirect k8s.io/klog/v2 v2.120.1 // indirect k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect diff --git a/backend/internal/coordinator/jobs.go b/backend/internal/coordinator/jobs.go index 65cad85..b9fa415 100644 --- a/backend/internal/coordinator/jobs.go +++ b/backend/internal/coordinator/jobs.go @@ -1,18 +1,135 @@ package coordinator import ( + "context" + cryptorand "crypto/rand" + "encoding/hex" "errors" "fmt" "strconv" "strings" + "time" 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" + "k8s.io/apimachinery/pkg/types" ) -const runnerContainerName = "pgbench-runner" +const ( + runnerContainerName = "pgbench-runner" + benchmarkJobNamePrefix = "plugin-bench-pgbench-" + credentialCleanupTimeout = 10 * time.Second +) + +// JobRef identifies the Job created for one benchmark run. +type JobRef struct { + Name string + UID types.UID +} + +type benchmarkResources struct { + secret SecretRef + job JobRef +} + +// createBenchmarkResources creates the Secret before the Job that consumes +// it. If Job creation fails, the Secret is removed before the error returns. +// Monitoring and normal run cleanup are handled by the execution lifecycle. +func (c *Coordinator) createBenchmarkResources(ctx context.Context, connection Connection, options Options, labels map[string]string) (benchmarkResources, error) { + var resources benchmarkResources + if c == nil { + return resources, errors.New("coordinator is nil") + } + if ctx == nil { + return resources, errors.New("resource creation context is nil") + } + if err := c.config.Validate(); err != nil { + return resources, fmt.Errorf("invalid coordinator configuration: %w", err) + } + if err := connection.Validate(); err != nil { + return resources, fmt.Errorf("invalid database connection: %w", err) + } + if err := options.Validate(); err != nil { + return resources, fmt.Errorf("invalid benchmark options: %w", err) + } + if _, err := resourceRequirements(c.config.Resources); err != nil { + return resources, fmt.Errorf("invalid runner resources: %w", err) + } + if err := ctx.Err(); err != nil { + return resources, err + } + + secret, err := c.createCredentialSecret(ctx, connection, labels) + if err != nil { + return resources, err + } + + job, err := c.createBenchmarkJob(ctx, secret, options, labels) + if err == nil { + return benchmarkResources{secret: secret, job: job}, nil + } + + // Cleanup must survive caller cancellation but still have a deadline. + cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), credentialCleanupTimeout) + defer cancel() + cleanupErr := c.deleteCredentialSecret(cleanupCtx, secret) + if cleanupErr != nil { + return resources, errors.Join(err, fmt.Errorf("cleanup credential Secret after Job creation failure: %w", cleanupErr)) + } + return resources, err +} + +// 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) { + var reference JobRef + if c == nil { + return reference, errors.New("coordinator is nil") + } + if ctx == nil { + return reference, errors.New("job creation context is nil") + } + if c.kubeClient == nil { + return reference, ErrKubernetesClientRequired + } + if err := c.config.Validate(); err != nil { + return reference, fmt.Errorf("invalid coordinator configuration: %w", err) + } + if strings.TrimSpace(secretRef.Name) == "" { + return reference, errors.New("credential Secret name is required") + } + if err := options.Validate(); err != nil { + return reference, fmt.Errorf("invalid benchmark options: %w", err) + } + if err := ctx.Err(); err != nil { + return reference, err + } + + name, err := generateBenchmarkJobName() + if err != nil { + return reference, err + } + job, err := newBenchmarkJob(c.config, name, secretRef.Name, options, labels) + if err != nil { + return reference, fmt.Errorf("build benchmark Job: %w", err) + } + + created, err := c.kubeClient.BatchV1().Jobs(c.config.WorkloadNamespace).Create(ctx, job, metav1.CreateOptions{}) + if err != nil { + return reference, fmt.Errorf("create benchmark Job: %w", err) + } + return JobRef{Name: created.Name, UID: created.UID}, nil +} + +func generateBenchmarkJobName() (string, error) { + var suffix [8]byte + if _, err := cryptorand.Read(suffix[:]); err != nil { + return "", fmt.Errorf("generate benchmark Job name: %w", err) + } + return benchmarkJobNamePrefix + hex.EncodeToString(suffix[:]), nil +} // newBenchmarkJob builds the per-run Job object. It only creates an in-memory // Kubernetes object; the caller is responsible for API calls and cleanup. diff --git a/backend/internal/coordinator/jobs_test.go b/backend/internal/coordinator/jobs_test.go index c0d29c1..55e5c46 100644 --- a/backend/internal/coordinator/jobs_test.go +++ b/backend/internal/coordinator/jobs_test.go @@ -1,12 +1,319 @@ package coordinator import ( + "context" + "errors" + "strings" "testing" + "time" + 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" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + ktesting "k8s.io/client-go/testing" ) +// The fake client does not expose request contexts through its reactors. +type cleanupContextClient struct { + kubernetes.Interface + check func(context.Context) +} + +func (c cleanupContextClient) CoreV1() typedcorev1.CoreV1Interface { + return cleanupContextCore{CoreV1Interface: c.Interface.CoreV1(), check: c.check} +} + +type cleanupContextCore struct { + typedcorev1.CoreV1Interface + check func(context.Context) +} + +func (c cleanupContextCore) Secrets(namespace string) typedcorev1.SecretInterface { + return cleanupContextSecrets{SecretInterface: c.CoreV1Interface.Secrets(namespace), check: c.check} +} + +type cleanupContextSecrets struct { + typedcorev1.SecretInterface + check func(context.Context) +} + +func (s cleanupContextSecrets) Delete(ctx context.Context, name string, options metav1.DeleteOptions) error { + s.check(ctx) + return s.SecretInterface.Delete(ctx, name, options) +} + +func TestResourceCleanupSurvivesCancellationWithDeadline(t *testing.T) { + client := fake.NewSimpleClientset() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + client.PrependReactor("create", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) { + action.(ktesting.CreateAction).GetObject().(*corev1.Secret).UID = "secret-uid" + cancel() + return false, nil, nil + }) + called := false + c, err := New(validConfig(), cleanupContextClient{Interface: client, check: func(cleanupCtx context.Context) { + called = true + if cleanupCtx.Err() != nil { + t.Fatal("cleanup inherited caller cancellation") + } + deadline, ok := cleanupCtx.Deadline() + if !ok || time.Until(deadline) <= 0 || time.Until(deadline) > credentialCleanupTimeout { + t.Fatal("cleanup must have a bounded future deadline") + } + }}) + if err != nil { + t.Fatal(err) + } + _, err = c.createBenchmarkResources(ctx, validConnection(), DefaultOptions(), nil) + if !errors.Is(err, context.Canceled) || !called { + t.Fatalf("error = %v, cleanup called = %v", err, called) + } + for _, action := range client.Actions() { + if action.Matches("create", "jobs") { + t.Fatal("created Job after cancellation") + } + } + secrets, err := client.CoreV1().Secrets(validConfig().WorkloadNamespace).List(context.Background(), metav1.ListOptions{}) + if err != nil || len(secrets.Items) != 0 { + t.Fatal("credential Secret was not cleaned up") + } +} + +func TestResourceCreationRejectsInvalidResourcesBeforeAPIWrite(t *testing.T) { + for _, resources := range []Resources{ + {CPURequest: "invalid"}, + {MemoryRequest: "-1Mi"}, + {CPURequest: "2", CPULimit: "1"}, + {CPURequest: "0.0001"}, + } { + client := fake.NewSimpleClientset() + config := validConfig() + config.Resources = resources + c, err := New(config, client) + if err != nil { + t.Fatal(err) + } + _, err = c.createBenchmarkResources(context.Background(), validConnection(), DefaultOptions(), nil) + if err == nil || len(client.Actions()) != 0 { + t.Fatalf("invalid resources must fail before API calls: error=%v actions=%v", err, client.Actions()) + } + } +} + +func TestResourceCreationPreservesJobAndCleanupErrors(t *testing.T) { + client := fake.NewSimpleClientset() + jobErr, cleanupErr := errors.New("job failed"), errors.New("cleanup failed") + client.PrependReactor("create", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) { + action.(ktesting.CreateAction).GetObject().(*corev1.Secret).UID = "secret-uid" + return false, nil, nil + }) + client.PrependReactor("create", "jobs", func(ktesting.Action) (bool, runtime.Object, error) { + return true, nil, jobErr + }) + client.PrependReactor("delete", "secrets", func(ktesting.Action) (bool, runtime.Object, error) { + return true, nil, cleanupErr + }) + c, err := New(validConfig(), client) + if err != nil { + t.Fatal(err) + } + _, err = c.createBenchmarkResources(context.Background(), validConnection(), DefaultOptions(), nil) + if !errors.Is(err, jobErr) || !errors.Is(err, cleanupErr) { + t.Fatalf("error = %v, want both causes", err) + } +} + +func TestSecretCreationFailurePreventsJobCreation(t *testing.T) { + client := fake.NewSimpleClientset() + cause := errors.New("secret creation failed") + client.PrependReactor("create", "secrets", func(ktesting.Action) (bool, runtime.Object, error) { + return true, nil, cause + }) + c, err := New(validConfig(), client) + if err != nil { + t.Fatal(err) + } + _, err = c.createBenchmarkResources(context.Background(), validConnection(), DefaultOptions(), nil) + if !errors.Is(err, cause) { + t.Fatalf("error = %v, want Secret failure", err) + } + if actions := client.Actions(); len(actions) != 1 || !actions[0].Matches("create", "secrets") { + t.Fatal("Secret creation failure must prevent further resource operations") + } +} + +func TestCreateBenchmarkJob(t *testing.T) { + client := fake.NewSimpleClientset() + client.PrependReactor("create", "jobs", func(action ktesting.Action) (bool, runtime.Object, error) { + job := action.(ktesting.CreateAction).GetObject().(*batchv1.Job) + job.UID = types.UID("job-uid-1") + return false, nil, nil + }) + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + jobRef, err := coordinator.createBenchmarkJob(context.Background(), SecretRef{Name: "plugin-bench-credentials-run-1"}, DefaultOptions(), nil) + if err != nil { + t.Fatalf("createBenchmarkJob() error = %v", err) + } + if !strings.HasPrefix(jobRef.Name, benchmarkJobNamePrefix) || jobRef.UID == "" { + t.Fatalf("JobRef = %+v, want generated name and UID", jobRef) + } + job, err := client.BatchV1().Jobs("plugin-bench-workloads").Get(context.Background(), jobRef.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("Get() error = %v", err) + } + container := job.Spec.Template.Spec.Containers[0] + for _, variable := range container.Env[:5] { + if variable.ValueFrom == nil || variable.ValueFrom.SecretKeyRef == nil || variable.ValueFrom.SecretKeyRef.Name != "plugin-bench-credentials-run-1" { + t.Fatalf("%s does not reference the created Secret", variable.Name) + } + } +} + +func TestCreateBenchmarkResourcesCreatesSecretBeforeJob(t *testing.T) { + client := fake.NewSimpleClientset() + var actions []string + client.PrependReactor("create", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) { + actions = append(actions, "secret") + secret := action.(ktesting.CreateAction).GetObject().(*corev1.Secret) + secret.UID = types.UID("secret-uid-1") + return false, nil, nil + }) + client.PrependReactor("create", "jobs", func(action ktesting.Action) (bool, runtime.Object, error) { + actions = append(actions, "job") + job := action.(ktesting.CreateAction).GetObject().(*batchv1.Job) + job.UID = types.UID("job-uid-1") + return false, nil, nil + }) + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + resources, err := coordinator.createBenchmarkResources(context.Background(), validConnection(), DefaultOptions(), nil) + if err != nil { + t.Fatalf("createBenchmarkResources() error = %v", err) + } + if !strings.HasPrefix(resources.secret.Name, credentialSecretNamePrefix) || resources.secret.UID == "" { + t.Fatalf("SecretRef = %+v, want generated name and UID", resources.secret) + } + if !strings.HasPrefix(resources.job.Name, benchmarkJobNamePrefix) || resources.job.UID == "" { + t.Fatalf("JobRef = %+v, want generated name and UID", resources.job) + } + if strings.Join(actions, ",") != "secret,job" { + t.Fatalf("resource creation order = %v, want [secret job]", actions) + } + job, err := client.BatchV1().Jobs(validConfig().WorkloadNamespace).Get(context.Background(), resources.job.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("Get(Job) error = %v", err) + } + 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) + } +} + +func TestCreateBenchmarkResourcesDeletesSecretWhenJobCreationFails(t *testing.T) { + client := fake.NewSimpleClientset() + client.PrependReactor("create", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) { + secret := action.(ktesting.CreateAction).GetObject().(*corev1.Secret) + secret.UID = types.UID("secret-uid-1") + return false, nil, nil + }) + client.PrependReactor("create", "jobs", func(ktesting.Action) (bool, runtime.Object, error) { + return true, nil, errors.New("Job API unavailable") + }) + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + _, err = coordinator.createBenchmarkResources(context.Background(), validConnection(), DefaultOptions(), nil) + creationErr := err + if creationErr == nil || !strings.Contains(creationErr.Error(), "Job API unavailable") { + t.Fatalf("error = %v, want Job API error", creationErr) + } + secrets, err := client.CoreV1().Secrets(validConfig().WorkloadNamespace).List(context.Background(), metav1.ListOptions{}) + if err != nil { + t.Fatalf("List(Secret) error = %v", err) + } + if len(secrets.Items) != 0 { + t.Fatalf("created %d Secret(s), want cleanup after Job failure", len(secrets.Items)) + } + if strings.Contains(creationErr.Error(), validConnection().Password) { + t.Fatal("resource creation error exposed database credentials") + } +} + +func TestCreateBenchmarkResourcesRejectsInvalidInputBeforeCreatingResources(t *testing.T) { + client := fake.NewSimpleClientset() + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + connection := validConnection() + connection.Database = "" + + _, err = coordinator.createBenchmarkResources(context.Background(), connection, DefaultOptions(), nil) + if err == nil || !strings.Contains(err.Error(), "database name is required") { + t.Fatalf("error = %v, want validation error", err) + } + secrets, err := client.CoreV1().Secrets(validConfig().WorkloadNamespace).List(context.Background(), metav1.ListOptions{}) + if err != nil { + t.Fatalf("List(Secret) error = %v", err) + } + jobs, err := client.BatchV1().Jobs(validConfig().WorkloadNamespace).List(context.Background(), metav1.ListOptions{}) + if err != nil { + t.Fatalf("List(Job) error = %v", err) + } + if len(secrets.Items) != 0 || len(jobs.Items) != 0 { + t.Fatalf("invalid input created resources: %d Secret(s), %d Job(s)", len(secrets.Items), len(jobs.Items)) + } +} + +func TestCreateBenchmarkJobRejectsCanceledContext(t *testing.T) { + client := fake.NewSimpleClientset() + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err = coordinator.createBenchmarkJob(ctx, SecretRef{Name: "credentials-run-1"}, DefaultOptions(), nil) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled", err) + } +} + +func TestCreateBenchmarkJobDoesNotExposeCredentialsOnAPIError(t *testing.T) { + client := fake.NewSimpleClientset() + client.PrependReactor("create", "jobs", func(ktesting.Action) (bool, runtime.Object, error) { + return true, nil, errors.New("Kubernetes API unavailable") + }) + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + _, err = coordinator.createBenchmarkJob(context.Background(), SecretRef{Name: "credentials-run-1"}, DefaultOptions(), nil) + if err == nil || !strings.Contains(err.Error(), "Kubernetes API unavailable") { + t.Fatalf("error = %v, want API error", err) + } + if strings.Contains(err.Error(), "sensitive-test-password") { + t.Fatal("API error exposed database credentials") + } +} + func TestNewBenchmarkJobBuildsRunnerManifest(t *testing.T) { labels := map[string]string{"plugin-bench/run": "run-1"} options := DefaultOptions() diff --git a/backend/internal/coordinator/secrets.go b/backend/internal/coordinator/secrets.go index d0d5cec..1b83534 100644 --- a/backend/internal/coordinator/secrets.go +++ b/backend/internal/coordinator/secrets.go @@ -1,21 +1,116 @@ package coordinator import ( + "context" + cryptorand "crypto/rand" + "encoding/hex" "errors" + "fmt" "strings" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" ) const ( - secretKeyHost = "PGHOST" - secretKeyPort = "PGPORT" - secretKeyDatabase = "PGDATABASE" - secretKeyUser = "PGUSER" - secretKeyPassword = "PGPASSWORD" + credentialSecretNamePrefix = "plugin-bench-credentials-" + secretKeyHost = "PGHOST" + secretKeyPort = "PGPORT" + secretKeyDatabase = "PGDATABASE" + secretKeyUser = "PGUSER" + secretKeyPassword = "PGPASSWORD" ) +// SecretRef identifies the credential Secret created for one benchmark run. +// The UID allows later cleanup to verify the exact resource being deleted. +type SecretRef struct { + Name string + UID types.UID +} + +// createCredentialSecret creates the per-run credential Secret in the +// configured workload namespace. It does not log or return the credentials. +func (c *Coordinator) createCredentialSecret(ctx context.Context, connection Connection, labels map[string]string) (SecretRef, error) { + var reference SecretRef + if c == nil { + return reference, errors.New("coordinator is nil") + } + if ctx == nil { + return reference, errors.New("secret creation context is nil") + } + if c.kubeClient == nil { + return reference, ErrKubernetesClientRequired + } + if err := c.config.Validate(); err != nil { + return reference, fmt.Errorf("invalid coordinator configuration: %w", err) + } + if err := ctx.Err(); err != nil { + return reference, err + } + name, err := generateCredentialSecretName() + if err != nil { + return reference, err + } + + secret, err := newCredentialSecret(c.config.WorkloadNamespace, name, connection, labels) + if err != nil { + return reference, fmt.Errorf("build credential Secret: %w", err) + } + + created, err := c.kubeClient.CoreV1().Secrets(c.config.WorkloadNamespace).Create(ctx, secret, metav1.CreateOptions{}) + if err != nil { + return reference, fmt.Errorf("create credential Secret: %w", err) + } + return SecretRef{Name: created.Name, UID: created.UID}, nil +} + +// deleteCredentialSecret removes a credential Secret created for a run. A +// missing Secret is already in the desired state and is treated as success. +func (c *Coordinator) deleteCredentialSecret(ctx context.Context, reference SecretRef) error { + if c == nil { + return errors.New("coordinator is nil") + } + if ctx == nil { + return errors.New("secret deletion context is nil") + } + if c.kubeClient == nil { + return ErrKubernetesClientRequired + } + if err := c.config.Validate(); err != nil { + return fmt.Errorf("invalid coordinator configuration: %w", err) + } + if strings.TrimSpace(reference.Name) == "" { + return errors.New("credential Secret name is required") + } + if reference.UID == "" { + return errors.New("credential Secret UID is required") + } + if err := ctx.Err(); err != nil { + return err + } + + err := c.kubeClient.CoreV1().Secrets(c.config.WorkloadNamespace).Delete(ctx, reference.Name, metav1.DeleteOptions{ + Preconditions: &metav1.Preconditions{UID: &reference.UID}, + }) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + return fmt.Errorf("delete credential Secret: %w", err) + } + return nil +} + +func generateCredentialSecretName() (string, error) { + var suffix [8]byte + if _, err := cryptorand.Read(suffix[:]); err != nil { + return "", fmt.Errorf("generate credential Secret name: %w", err) + } + return credentialSecretNamePrefix + hex.EncodeToString(suffix[:]), nil +} + // newCredentialSecret builds the per-run credential Secret. It only creates // an in-memory Kubernetes object; the caller is responsible for API calls and // cleanup. diff --git a/backend/internal/coordinator/secrets_test.go b/backend/internal/coordinator/secrets_test.go index 0482f57..86389f6 100644 --- a/backend/internal/coordinator/secrets_test.go +++ b/backend/internal/coordinator/secrets_test.go @@ -1,8 +1,18 @@ package coordinator import ( + "context" + "errors" "strings" "testing" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/fake" + ktesting "k8s.io/client-go/testing" ) func TestNewCredentialSecretRejectsInvalidInput(t *testing.T) { @@ -64,3 +74,139 @@ func TestNewCredentialSecretBuildsConnectionData(t *testing.T) { t.Fatal("secret does not contain the expected connection data") } } + +func TestCreateCredentialSecret(t *testing.T) { + client := fake.NewSimpleClientset() + client.PrependReactor("create", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) { + secret := action.(ktesting.CreateAction).GetObject().(*corev1.Secret) + secret.UID = types.UID("uid-run-1") + return false, nil, nil + }) + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + connection := Connection{ + Host: "postgres.example.com", + Port: 5432, + Database: "pgbench", + Username: "postgres", + Password: "sensitive-test-password", + } + + reference, err := coordinator.createCredentialSecret(context.Background(), connection, map[string]string{"plugin-bench/run-id": "run-1"}) + if err != nil { + t.Fatalf("createCredentialSecret() error = %v", err) + } + if !strings.HasPrefix(reference.Name, credentialSecretNamePrefix) || reference.UID == "" { + t.Fatalf("SecretRef = %+v, want generated name and UID", reference) + } + created, err := client.CoreV1().Secrets("plugin-bench-workloads").Get(context.Background(), reference.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("Get() error = %v", err) + } + if string(created.Data[secretKeyPassword]) != connection.Password { + t.Fatal("created Secret does not contain the password") + } + if created.Labels["plugin-bench/run-id"] != "run-1" { + t.Fatal("created Secret did not retain run labels") + } +} + +func TestCreateCredentialSecretRejectsCanceledContext(t *testing.T) { + client := fake.NewSimpleClientset() + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err = coordinator.createCredentialSecret(ctx, validConnection(), nil) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled", err) + } + secrets, err := client.CoreV1().Secrets("plugin-bench-workloads").List(context.Background(), metav1.ListOptions{}) + if err != nil { + t.Fatalf("List() error = %v", err) + } + if len(secrets.Items) != 0 { + t.Fatalf("created %d Secret(s) after cancellation, want none", len(secrets.Items)) + } +} + +func TestCreateCredentialSecretDoesNotExposeCredentialsOnAPIError(t *testing.T) { + client := fake.NewSimpleClientset() + client.PrependReactor("create", "secrets", func(ktesting.Action) (bool, runtime.Object, error) { + return true, nil, errors.New("Kubernetes API unavailable") + }) + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + _, err = coordinator.createCredentialSecret(context.Background(), validConnection(), nil) + if err == nil || !strings.Contains(err.Error(), "Kubernetes API unavailable") { + t.Fatalf("error = %v, want API error", err) + } + if strings.Contains(err.Error(), "sensitive-test-password") { + t.Fatal("API error exposed database credentials") + } +} + +func TestDeleteCredentialSecret(t *testing.T) { + client := fake.NewSimpleClientset(&corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "credentials-run-1", Namespace: "plugin-bench-workloads", UID: "secret-uid-1"}, + }) + coordinator, err := New(validConfig(), client) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + client.PrependReactor("delete", "secrets", func(action ktesting.Action) (bool, runtime.Object, error) { + options := action.(ktesting.DeleteAction).GetDeleteOptions() + if options.Preconditions == nil || options.Preconditions.UID == nil || *options.Preconditions.UID != "secret-uid-1" { + t.Fatal("deletion must specify the recorded Secret UID") + } + return false, nil, nil + }) + if err := coordinator.deleteCredentialSecret(context.Background(), SecretRef{Name: "credentials-run-1", UID: "secret-uid-1"}); err != nil { + t.Fatalf("deleteCredentialSecret() error = %v", err) + } + _, err = client.CoreV1().Secrets(validConfig().WorkloadNamespace).Get(context.Background(), "credentials-run-1", metav1.GetOptions{}) + if !apierrors.IsNotFound(err) { + t.Fatalf("Get() error = %v, want NotFound", err) + } +} + +func validConnection() Connection { + return Connection{ + Host: "postgres.example.com", + Port: 5432, + Database: "pgbench", + Username: "postgres", + Password: "sensitive-test-password", + } +} + +func TestDeleteCredentialSecretRequiresUID(t *testing.T) { + client := fake.NewSimpleClientset() + c, err := New(validConfig(), client) + if err != nil { + t.Fatal(err) + } + err = c.deleteCredentialSecret(context.Background(), SecretRef{Name: "credentials"}) + if err == nil || len(client.Actions()) != 0 { + t.Fatal("missing UID must prevent deletion") + } +} + +func TestDeleteCredentialSecretAlreadyAbsent(t *testing.T) { + c, err := New(validConfig(), fake.NewSimpleClientset()) + if err != nil { + t.Fatal(err) + } + if err := c.deleteCredentialSecret(context.Background(), SecretRef{Name: "credentials", UID: "original"}); err != nil { + t.Fatalf("absent Secret should count as cleaned up: %v", err) + } +}