diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 991c4ba..0762b61 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,6 +39,67 @@ jobs: - run: cd backend && go test ./... - run: cd backend && go build -o /dev/null . + runner: + name: pgbench Runner + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0 + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0 + with: + go-version: "1.22" + cache-dependency-path: runners/pgbench/go.mod + - name: Vet runner + run: cd runners/pgbench && go vet ./... + - name: Test runner + run: cd runners/pgbench && go test -race ./... + - name: Build runner image + run: docker build -t plugin-bench-pgbench:ci runners/pgbench + - name: Integration test runner + timeout-minutes: 3 + shell: bash + run: | + set -euo pipefail + docker network create pgbench-ci + docker run -d --name pgbench-ci-db --network pgbench-ci \ + -e POSTGRES_PASSWORD=ci-password -e POSTGRES_DB=pgbench postgres:16-alpine + ready=false + for attempt in {1..30}; do + if docker exec pgbench-ci-db pg_isready -h 127.0.0.1 -U postgres -d pgbench; then + ready=true + break + fi + sleep 1 + done + if [ "$ready" != true ]; then + docker logs pgbench-ci-db + exit 1 + fi + runner=(docker run --rm --network pgbench-ci + -e PGHOST=pgbench-ci-db -e PGDATABASE=pgbench -e PGUSER=postgres + -e PGCONNECT_TIMEOUT=3 -e BENCH_DURATION=2) + for initialize in true false; do + "${runner[@]}" -e PGPASSWORD=ci-password -e BENCH_INITIALIZE="$initialize" \ + plugin-bench-pgbench:ci > benchmark.log 2>&1 + cat benchmark.log + grep -Eq '^tps = [0-9]' benchmark.log + grep -Eq '^latency average = [0-9]' benchmark.log + done + if "${runner[@]}" -e PGPASSWORD=wrong-ci-password \ + plugin-bench-pgbench:ci > failure.log 2>&1; then + echo "Expected authentication failure" + exit 1 + fi + grep -q 'password authentication failed' failure.log + if grep -Fq 'wrong-ci-password' failure.log; then + echo "Failure output exposed the password" + exit 1 + fi + - name: Clean up integration resources + if: always() + run: | + docker rm -fv pgbench-ci-db || true + docker network rm pgbench-ci || true + helm-lint: name: Helm lint runs-on: ubuntu-latest diff --git a/runners/pgbench/Dockerfile b/runners/pgbench/Dockerfile new file mode 100644 index 0000000..92b861d --- /dev/null +++ b/runners/pgbench/Dockerfile @@ -0,0 +1,26 @@ +FROM golang:1.22-alpine AS builder + +WORKDIR /src + +COPY go.mod ./ +COPY main.go ./ + +RUN CGO_ENABLED=0 GOOS=linux go build \ + -trimpath \ + -ldflags="-s -w" \ + -o /out/pgbench-runner \ + . + +FROM postgres:16-alpine + +# pgbench is provided by PostgreSQL's contrib package rather than the server +# package. Verify it during the image build so the runner cannot start without +# its benchmark executable. +RUN apk add --no-cache postgresql16-contrib \ + && pgbench --version + +COPY --from=builder --chown=postgres:postgres /out/pgbench-runner /usr/local/bin/pgbench-runner + +USER postgres + +ENTRYPOINT ["/usr/local/bin/pgbench-runner"] diff --git a/runners/pgbench/README.md b/runners/pgbench/README.md new file mode 100644 index 0000000..17770ad --- /dev/null +++ b/runners/pgbench/README.md @@ -0,0 +1,93 @@ +# pgbench Runner + +The runner connects to PostgreSQL, optionally initializes pgbench tables, and prints native pgbench results. Results are not stored yet; the coordinator will later run this image as a Kubernetes Job. + +## Build + +From the repository root: + +```bash +docker build -t plugin-bench-pgbench:dev runners/pgbench +``` + +## Disposable PostgreSQL database + +```bash +docker network create pgbench-test +docker run -d \ + --name pgbench-db-test \ + --network pgbench-test \ + -e POSTGRES_PASSWORD=postgres \ + -e POSTGRES_DB=pgbench \ + postgres:16-alpine + +ready=false +for attempt in {1..30}; do + if docker exec pgbench-db-test pg_isready -U postgres -d pgbench; then + ready=true + break + fi + sleep 1 +done +if [ "$ready" != true ]; then + echo "PostgreSQL did not become ready" + exit 1 +fi +``` + +## Configuration + +Required connection variables: + +```text +PGHOST PostgreSQL hostname +PGDATABASE Database name +PGUSER PostgreSQL user +PGPASSWORD Password, or use PGPASSFILE +``` + +Optional connection variables include `PGPORT` (default `5432`), `PGCONNECT_TIMEOUT` (default `10`), `PGSSLMODE`, and `PGSSLROOTCERT`. + +For `PGPASSFILE`, mount a password file readable by the container's `postgres` user, with mode `0600`, and supply its container path. When set, `PGPASSWORD` takes precedence. + +Benchmark variables: + +```text +BENCH_INITIALIZE Initialize tables; default false +BENCH_DURATION Duration in seconds; default 60 +BENCH_CLIENTS Concurrent clients; default 1 +BENCH_THREADS pgbench threads; default 1 +BENCH_SCALE Initialization scale; default 1 +``` + +`BENCH_THREADS` cannot exceed `BENCH_CLIENTS`. + +## Run + +Initialize the disposable database and run a 30-second benchmark. `BENCH_INITIALIZE=true` drops and recreates existing pgbench tables, so use it only with a disposable database: + +```bash +docker run --rm \ + --network pgbench-test \ + -e PGHOST=pgbench-db-test \ + -e PGPORT=5432 \ + -e PGDATABASE=pgbench \ + -e PGUSER=postgres \ + -e PGPASSWORD=postgres \ + -e BENCH_INITIALIZE=true \ + -e BENCH_DURATION=30 \ + -e BENCH_CLIENTS=2 \ + -e BENCH_THREADS=2 \ + plugin-bench-pgbench:dev +``` + +The output includes `tps`, average latency, and failed transactions. Run again without `BENCH_INITIALIZE` to reuse the tables. + +The runner passes connection settings through PostgreSQL environment variables and does not put credentials in command-line arguments. It returns a non-zero exit code for invalid configuration, connection failures, initialization failures, or benchmark failures. + +## Cleanup + +```bash +docker rm -f pgbench-db-test +docker network rm pgbench-test +``` diff --git a/runners/pgbench/go.mod b/runners/pgbench/go.mod new file mode 100644 index 0000000..3204be6 --- /dev/null +++ b/runners/pgbench/go.mod @@ -0,0 +1,3 @@ +module github.com/openeverest/plugin-bench/runners/pgbench + +go 1.22 diff --git a/runners/pgbench/main.go b/runners/pgbench/main.go new file mode 100644 index 0000000..5b64100 --- /dev/null +++ b/runners/pgbench/main.go @@ -0,0 +1,153 @@ +package main + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "os/signal" + "strconv" + "strings" + "syscall" + "time" +) + +type config struct { + port, timeout, duration, clients, threads, scale int + initialize bool +} + +func loadConfig(getenv func(string) string) (config, error) { + var c config + for _, name := range []string{"PGHOST", "PGDATABASE", "PGUSER"} { + if strings.TrimSpace(getenv(name)) == "" { + return c, fmt.Errorf("%s is required", name) + } + } + if getenv("PGPASSWORD") == "" && getenv("PGPASSFILE") == "" { + return c, errors.New("PGPASSWORD or PGPASSFILE is required") + } + if getenv("PGPASSWORD") == "" { + path := getenv("PGPASSFILE") + info, err := os.Stat(path) + if err != nil || !info.Mode().IsRegular() { + return c, errors.New("PGPASSFILE must refer to an existing regular file") + } + // libpq ignores password files accessible by group or other users. + if info.Mode().Perm()&0077 != 0 { + return c, errors.New("PGPASSFILE must not allow group or other access (use mode 0600)") + } + file, err := os.Open(path) + if err != nil { + return c, errors.New("PGPASSFILE must be readable by the runner user") + } + file.Close() + } + for _, field := range []struct { + name string + fallback, maximum int + target *int + }{ + {"PGPORT", 5432, 65535, &c.port}, + {"PGCONNECT_TIMEOUT", 10, 2147483647, &c.timeout}, + {"BENCH_DURATION", 60, 2147483647, &c.duration}, + {"BENCH_CLIENTS", 1, 2147483647, &c.clients}, + {"BENCH_THREADS", 1, 2147483647, &c.threads}, + {"BENCH_SCALE", 1, 2147483647, &c.scale}, + } { + value := getenv(field.name) + if value == "" { + *field.target = field.fallback + continue + } + n, err := strconv.ParseUint(value, 10, 31) + if err != nil || n == 0 || n > uint64(field.maximum) { + return c, fmt.Errorf("%s must be an integer between 1 and %d", field.name, field.maximum) + } + *field.target = int(n) + } + switch getenv("BENCH_INITIALIZE") { + case "", "false": + case "true": + c.initialize = true + default: + return c, errors.New("BENCH_INITIALIZE must be true or false") + } + if c.threads > c.clients { + return c, errors.New("BENCH_THREADS cannot exceed BENCH_CLIENTS") + } + return c, nil +} + +// commandEnvironment preserves connection and TLS settings, replacing defaulted +// values without putting credentials into command arguments. +func commandEnvironment(env []string, c config) []string { + result := make([]string, 0, len(env)+2) + for _, entry := range env { + if !strings.HasPrefix(entry, "PGPORT=") && !strings.HasPrefix(entry, "PGCONNECT_TIMEOUT=") { + result = append(result, entry) + } + } + return append(result, "PGPORT="+strconv.Itoa(c.port), "PGCONNECT_TIMEOUT="+strconv.Itoa(c.timeout)) +} + +// execute streams pgbench output and terminates the child when the runner is +// cancelled. A child that ignores SIGTERM is killed after five seconds. +func execute(ctx context.Context, executable string, args, env []string, stdout, stderr io.Writer) error { + cmd := exec.CommandContext(ctx, executable, args...) + cmd.Env = env + cmd.Stdout, cmd.Stderr = stdout, stderr + cmd.Cancel = func() error { return cmd.Process.Signal(syscall.SIGTERM) } + cmd.WaitDelay = 5 * time.Second + return cmd.Run() +} + +func run(ctx context.Context, c config, invoke func(context.Context, []string) error) error { + if c.initialize { + if err := invoke(ctx, []string{"-i", "-s", strconv.Itoa(c.scale)}); err != nil { + return fmt.Errorf("initialization failed: %w", err) + } + } + if err := ctx.Err(); err != nil { + return err + } + return invoke(ctx, []string{"-c", strconv.Itoa(c.clients), "-j", strconv.Itoa(c.threads), "-T", strconv.Itoa(c.duration)}) +} + +func runnerMain() int { + c, err := loadConfig(os.Getenv) + if err != nil { + fmt.Fprintln(os.Stderr, "configuration error:", err) + return 1 + } + executable, err := exec.LookPath("pgbench") + if err != nil { + fmt.Fprintln(os.Stderr, "runner error: pgbench executable was not found") + return 1 + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + env := commandEnvironment(os.Environ(), c) + err = run(ctx, c, func(ctx context.Context, args []string) error { + return execute(ctx, executable, args, env, os.Stdout, os.Stderr) + }) + if ctx.Err() != nil { + fmt.Fprintln(os.Stderr, "runner cancelled") + return 1 + } + if err != nil { + fmt.Fprintln(os.Stderr, "runner error: pgbench execution failed") + var exitErr *exec.ExitError + if errors.As(err, &exitErr) && exitErr.ExitCode() > 0 { + return exitErr.ExitCode() + } + return 1 + } + return 0 +} + +func main() { + os.Exit(runnerMain()) +} diff --git a/runners/pgbench/main_test.go b/runners/pgbench/main_test.go new file mode 100644 index 0000000..25386e1 --- /dev/null +++ b/runners/pgbench/main_test.go @@ -0,0 +1,162 @@ +package main + +import ( + "bytes" + "context" + "errors" + "os" + "os/exec" + "path/filepath" + "reflect" + "strings" + "testing" + "time" +) + +func validEnvironment() map[string]string { + return map[string]string{"PGHOST": "postgres", "PGDATABASE": "bench", "PGUSER": "postgres", "PGPASSWORD": "dummy-secret"} +} + +func TestConfig(t *testing.T) { + env := validEnvironment() + c, err := loadConfig(func(key string) string { return env[key] }) + if err != nil || c != (config{port: 5432, timeout: 10, duration: 60, clients: 1, threads: 1, scale: 1}) { + t.Fatalf("unexpected defaults: %+v, %v", c, err) + } + for _, field := range []string{"PGPORT", "PGCONNECT_TIMEOUT", "BENCH_DURATION", "BENCH_CLIENTS", "BENCH_THREADS", "BENCH_SCALE"} { + for _, value := range []string{"0", "00", "-1", "abc", "999999999999999999999999"} { + t.Run(field+"/"+value, func(t *testing.T) { + env := validEnvironment() + env[field] = value + _, err := loadConfig(func(key string) string { return env[key] }) + if err == nil || !strings.Contains(err.Error(), field) { + t.Fatalf("expected safe field error, got %v", err) + } + }) + } + } + for _, field := range []string{"PGHOST", "PGDATABASE", "PGUSER", "PGPASSWORD"} { + t.Run("missing/"+field, func(t *testing.T) { + env := validEnvironment() + delete(env, field) + _, err := loadConfig(func(key string) string { return env[key] }) + if err == nil || strings.Contains(err.Error(), "dummy-secret") { + t.Fatalf("expected safe validation error, got %v", err) + } + }) + } + for key, value := range map[string]string{"PGPORT": "65536", "BENCH_INITIALIZE": "yes", "BENCH_THREADS": "2"} { + env := validEnvironment() + env[key] = value + if _, err := loadConfig(func(key string) string { return env[key] }); err == nil { + t.Fatalf("accepted invalid %s", key) + } + } +} + +func TestPasswordFile(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "pgpass") + if err := os.WriteFile(path, []byte("postgres:5432:bench:postgres:dummy-secret\n"), 0600); err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + name, path string + mode os.FileMode + password string + wantError bool + }{ + {"valid", path, 0600, "", false}, + {"missing", filepath.Join(dir, "missing"), 0600, "", true}, + {"directory", dir, 0600, "", true}, + {"insecure permissions", path, 0644, "", true}, + {"unreadable", path, 0000, "", true}, + {"password takes precedence", filepath.Join(dir, "missing"), 0600, "dummy-secret", false}, + } { + t.Run(tc.name, func(t *testing.T) { + if tc.name == "unreadable" && os.Geteuid() == 0 { + t.Skip("root can read mode-0000 files") + } + if err := os.Chmod(path, tc.mode); err != nil { + t.Fatal(err) + } + env := validEnvironment() + env["PGPASSWORD"], env["PGPASSFILE"] = tc.password, tc.path + _, err := loadConfig(func(key string) string { return env[key] }) + if (err != nil) != tc.wantError { + t.Fatalf("error=%v, wantError=%v", err, tc.wantError) + } + if err != nil && (strings.Contains(err.Error(), tc.path) || strings.Contains(err.Error(), "dummy-secret")) { + t.Fatal("error exposed password file path or contents") + } + }) + } +} + +func TestRun(t *testing.T) { + failure := errors.New("child failure") + for _, tc := range []struct { + name string + initialize bool + failAt int + wantCalls int + }{ + {"benchmark only", false, 0, 1}, + {"initialize first", true, 0, 2}, + {"initialization failure", true, 1, 1}, + {"benchmark failure", false, 1, 1}, + } { + t.Run(tc.name, func(t *testing.T) { + var calls [][]string + err := run(context.Background(), config{initialize: tc.initialize, scale: 3, clients: 4, threads: 2, duration: 7}, + func(_ context.Context, args []string) error { + calls = append(calls, args) + if len(calls) == tc.failAt { + return failure + } + return nil + }) + if len(calls) != tc.wantCalls || errors.Is(err, failure) != (tc.failAt != 0) { + t.Fatalf("calls=%v, error=%v", calls, err) + } + want := [][]string{{"-c", "4", "-j", "2", "-T", "7"}} + if tc.initialize { + want = append([][]string{{"-i", "-s", "3"}}, want...) + } + if !reflect.DeepEqual(calls, want[:tc.wantCalls]) { + t.Fatalf("wrong arguments: %v", calls) + } + }) + } +} + +func TestChildProcess(t *testing.T) { + switch os.Getenv("RUNNER_TEST_CHILD") { + case "exit": + os.Stdout.WriteString(os.Getenv("PGPORT") + "/" + os.Getenv("PGCONNECT_TIMEOUT") + "/" + os.Getenv("PGSSLMODE")) + os.Exit(7) + case "wait": + time.Sleep(time.Minute) + os.Exit(0) + } +} + +func TestExecute(t *testing.T) { + env := commandEnvironment([]string{"RUNNER_TEST_CHILD=exit", "PGPORT=", "PGCONNECT_TIMEOUT=", "PGSSLMODE=require"}, config{port: 5432, timeout: 10}) + var output bytes.Buffer + err := execute(context.Background(), os.Args[0], []string{"-test.run=^TestChildProcess$"}, env, &output, &output) + var exitErr *exec.ExitError + if !errors.As(err, &exitErr) || exitErr.ExitCode() != 7 { + t.Fatalf("lost exit status: %v", err) + } + if output.String() != "5432/10/require" { + t.Fatalf("unexpected child environment: %q", output.String()) + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + start := time.Now() + err = execute(ctx, os.Args[0], []string{"-test.run=^TestChildProcess$"}, []string{"RUNNER_TEST_CHILD=wait"}, &output, &output) + if err == nil || ctx.Err() == nil || time.Since(start) > 10*time.Second { + t.Fatalf("child did not stop promptly: %v", err) + } +}