diff --git a/.github/workflows/ec2-leg.yml b/.github/workflows/ec2-leg.yml index fa3851713..f5f958178 100644 --- a/.github/workflows/ec2-leg.yml +++ b/.github/workflows/ec2-leg.yml @@ -5,6 +5,7 @@ name: Perf-eval EC2 leg (ephemeral) # Does not post to the PR -- the coordinator owns human-facing output. # Box bootstrap: perf-eval/bootstrap-common.sh + the leg script (inputs.leg_script); # runner-side polling: perf-eval/gather. +# Legs can chain via keep_instance + adopt_run_label + the instance_ip output. on: # workflow_call is the only trigger: a leg runs through the coordinator. @@ -62,6 +63,20 @@ on: required: false type: string default: '' + keep_instance: + description: 'Leave a passing leg''s box running for a chained successor; the box then owns its shutdown. A failing leg terminates as usual.' + required: false + type: boolean + default: false + adopt_run_label: + description: 'run_label of a prior keep_instance leg whose box this leg also terminates when it passes' + required: false + type: string + default: '' + outputs: + instance_ip: + description: 'Private IPv4 of the launched box, for chained legs' + value: ${{ jobs.ec2-leg.outputs.instance_ip }} secrets: AWS_GHA_ROLE_ARN: description: 'OIDC role ARN to assume for EC2 + S3 access' @@ -75,6 +90,8 @@ permissions: jobs: ec2-leg: name: Launch + await ephemeral perf-eval box + outputs: + instance_ip: ${{ steps.launch.outputs.instance_ip }} # Non-cancelling group keyed on the test commit SHA + leg. An instance can run # until the box’s self-terminate ceiling, so new runs queue. concurrency: @@ -149,7 +166,7 @@ jobs: echo '#!/usr/bin/env bash' echo "export TARGET_SHA=${{ steps.resolve.outputs.sha }} RUN_ID=${{ github.run_id }}-${{ github.run_attempt }}" echo "export BUCKET=$BUCKET RESULT_KEY=$RESULT_KEY" - echo "export SELF_TERMINATE_MINUTES=$SELF_TERMINATE_MINUTES" + echo "export SELF_TERMINATE_MINUTES=$SELF_TERMINATE_MINUTES BUDGET_MINUTES=${{ inputs.budget_minutes }}" if [ -n "$EXTRA_ENV" ]; then printf '%s\n' "$EXTRA_ENV"; fi cat "$PERF_EVAL_DIR/bootstrap-common.sh" cat "$LEG_SCRIPT" @@ -190,6 +207,7 @@ jobs: INSTANCE_ID=$(printf '%s' "$RUN_INSTANCES_JSON" | jq -r '.Instances[0].InstanceId') echo "instance_id=$INSTANCE_ID" >> "$GITHUB_OUTPUT" + echo "instance_ip=$(printf '%s' "$RUN_INSTANCES_JSON" | jq -r '.Instances[0].PrivateIpAddress')" >> "$GITHUB_OUTPUT" - name: Wait for SSM agent to register env: @@ -257,8 +275,29 @@ jobs: exit 1 fi + # The adopted box is tag-discovered. Adoption fires only on a pass. Note that + # a failed leg leaves the box to its serve ceiling so a re-run can reuse it. - name: Terminate instance - if: always() && steps.launch.outputs.instance_id != '' + if: always() + env: + INSTANCE_ID: ${{ steps.launch.outputs.instance_id }} + KEEP_INSTANCE: ${{ inputs.keep_instance }} + LEG_PASSED: ${{ steps.results.outputs.passed }} + ADOPT_RUN_LABEL: ${{ inputs.adopt_run_label }} run: | - aws ec2 terminate-instances \ - --instance-ids ${{ steps.launch.outputs.instance_id }} || true + if [ "$KEEP_INSTANCE" = "true" ] && [ "$LEG_PASSED" = "true" ]; then + echo "keep_instance: leaving $INSTANCE_ID running for the chained leg" + elif [ -n "$INSTANCE_ID" ]; then + aws ec2 terminate-instances --instance-ids "$INSTANCE_ID" || true + fi + if [ -n "$ADOPT_RUN_LABEL" ] && [ "$LEG_PASSED" = "true" ]; then + ADOPTED=$(aws ec2 describe-instances \ + --filters "Name=tag:run-id,Values=${{ github.run_id }}" \ + "Name=tag:Name,Values=perf-eval-$ADOPT_RUN_LABEL" \ + "Name=instance-state-name,Values=pending,running" \ + --query 'Reservations[].Instances[].InstanceId' --output text || true) + if [ -n "$ADOPTED" ]; then + echo "terminating adopted box(es): $ADOPTED" + aws ec2 terminate-instances --instance-ids $ADOPTED || true + fi + fi diff --git a/.github/workflows/load-test-coordinator.yml b/.github/workflows/load-test-coordinator.yml index 1b7240e01..96e28206c 100644 --- a/.github/workflows/load-test-coordinator.yml +++ b/.github/workflows/load-test-coordinator.yml @@ -46,6 +46,7 @@ jobs: target_sha: ${{ steps.ctx.outputs.target_sha }} pr_number: ${{ steps.ctx.outputs.pr_number }} legs: ${{ steps.ctx.outputs.legs }} + matrix_legs: ${{ steps.ctx.outputs.matrix_legs }} steps: - uses: actions/checkout@v4 with: @@ -71,11 +72,15 @@ jobs: [ -n "$BASELINE" ] || echo "::warning::no baseline release found; the go-bench leg will fail" echo "go-bench baseline: ${BASELINE:-}" + # One roster (comment order) drives both the report job and the + # matrix fan-out: entries carrying a script ride the matrix; the + # chained backfill + blaster pair runs as explicit jobs below. # Unquoted heredoc so $BASELINE expands into the go-bench entry; LEGS="$(jq -c . <> "$GITHUB_OUTPUT" - # The fan-out: one matrix entry per leg. + # The fan-out: one matrix entry per independent leg. load-test: name: ${{ matrix.leg.label }} needs: plan strategy: fail-fast: false # stops one leg's failure from cancelling other legs matrix: - leg: ${{ fromJSON(needs.plan.outputs.legs) }} + leg: ${{ fromJSON(needs.plan.outputs.matrix_legs) }} uses: ./.github/workflows/ec2-leg.yml secrets: inherit with: @@ -105,9 +111,41 @@ jobs: root_volume_gb: ${{ matrix.leg.root_volume_gb || 500 }} extra_env: ${{ matrix.leg.extra_env || '' }} + # Chained pair, serving side (keep box's RPC serving for the blaster leg). + backfill: + name: Backfill ingestion + needs: plan + uses: ./.github/workflows/ec2-leg.yml + secrets: inherit + with: + target_ref: ${{ needs.plan.outputs.target_sha }} + run_label: backfill + leg_script: cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh + budget_minutes: 345 + keep_instance: true + extra_env: | + export SERVE_AFTER_BACKFILL=true + + # Chained pair, blasting side: starts once backfill's leg passes (its box then + # enters catchup), probes the handed-off address until healthy, then blasts. + blaster: + name: Endpoint load test + needs: [plan, backfill] + uses: ./.github/workflows/ec2-leg.yml + secrets: inherit + with: + target_ref: ${{ needs.plan.outputs.target_sha }} + run_label: blaster + leg_script: cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/run-blaster.sh + budget_minutes: 255 + root_volume_gb: 50 + adopt_run_label: backfill + extra_env: | + export TARGET_RPC=http://${{ needs.backfill.outputs.instance_ip }}:8000 + report: name: Aggregate + report - needs: [plan, load-test] + needs: [plan, load-test, backfill, blaster] if: ${{ !cancelled() }} runs-on: ubuntu-latest timeout-minutes: 10 @@ -178,13 +216,14 @@ jobs: fi fi - # Authoritative gate: reflects the load-test legs, independent of whether - # the comment posted. - name: Set overall status if: always() + env: + NEEDS: ${{ toJSON(needs) }} run: | - if [ "${{ needs.load-test.result }}" != "success" ]; then - echo "::error::one or more legs did not pass (aggregate result: ${{ needs.load-test.result }})" + FAILED=$(jq -r '[to_entries[] | select(.value.result != "success") | .key] | join(", ")' <<<"$NEEDS") + if [ -n "$FAILED" ]; then + echo "::error::jobs did not pass: $FAILED" exit 1 fi echo "all load-test legs passed" diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh index 2193e35f8..f1315754a 100644 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh @@ -1,7 +1,9 @@ -# Backfill ingestion leg. Relies on bootstrap-common.sh's env, helpers, -# bootstrap_box, and run_leg. It hands off to `runner instantiate` to build -# stellar-rpc and time a backfill run against the datalake. The other half, `runner -# gather`, polls S3 for the result object. +# Backfill ingestion leg. Concatenated after bootstrap-common.sh in the rendered +# EC2 user-data, so it relies on that file's env, helpers, bootstrap_box, and +# run_leg. It hands off to `runner instantiate`, which builds stellar-rpc and +# times a backfill run against the pubnet datastore. The other half, `runner +# gather`, polls S3 for the result object. With SERVE_AFTER_BACKFILL=true the +# box then keeps serving for the chained blaster leg. LEG_TITLE="Backfill ingestion" bootstrap_box diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/instantiate.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/instantiate.go index 39d1a0fc9..acdb6869d 100644 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/instantiate.go +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/instantiate.go @@ -13,6 +13,7 @@ import ( "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/caarlos0/env/v11" "github.com/stellar/go-stellar-sdk/ingest/ledgerbackend" @@ -30,94 +31,124 @@ const ledgerThreshold = 384 // mirrors ingest.ledgerThreshold in backfill.go // backfillDoneRe matches the terminal line emitted on backfill's completion var backfillDoneRe = regexp.MustCompile(`Backfill process complete, ledgers \[(\d+) -> (\d+)\]`) -// instantiate is the instance's backfill task: it fetches + builds test fixtures, -// runs a timed backfill, then publishes the verdict. +// backfillEnv is the leg's env-derived config. +type backfillEnv struct { + Bucket string `env:"BUCKET" envDefault:"stellar-rpc-ci-load-test"` + Region string `env:"REGION" envDefault:"us-east-1"` + WorkDir string `env:"WORK_DIR" envDefault:"/data"` + ResultsFile string `env:"RESULTS_FILE" envDefault:"/tmp/results.md"` + ResultKey string `env:"RESULT_KEY"` + TargetSHA string `env:"TARGET_SHA"` + RunID string `env:"RUN_ID" envDefault:"manual"` + // ~1 day by default for cheap test runs; the full week is 120960. + Retention int `env:"HISTORY_RETENTION_WINDOW" envDefault:"17280"` + Deadline time.Duration `env:"BACKFILL_DEADLINE" envDefault:"4h"` + // serve on a non-loopback bind after the backfill completes + ServeAfter bool `env:"SERVE_AFTER_BACKFILL"` +} + +// instantiate fetches + builds test fixtures + runs a timed backfill, then +// publishes the verdict. With SERVE_AFTER_BACKFILL it keeps the daemon serving. func instantiate(ctx context.Context) error { - var ( - env = harness.GetEnv() - // ~1 day by default for cheap test runs; the full week is 120960. - retention = harness.Env("HISTORY_RETENTION_WINDOW", "17280") - deadline = harness.Env("BACKFILL_DEADLINE", "4h") - - binaryPath = filepath.Join(env["WORK_DIR"], "stellar-rpc-bin") // built here (the repo checkout is in WORK_DIR) - ) + cfg, err := env.ParseAs[backfillEnv]() + if err != nil { + return fmt.Errorf("parsing env: %w", err) // run_leg publishes the generic fail + } + binaryPath := filepath.Join(cfg.WorkDir, "stellar-rpc-bin") // built here (the repo checkout is in WORK_DIR) + retention := strconv.Itoa(cfg.Retention) // config template + report take it as a string + repoRoot, err := os.Getwd() if err != nil { return err } bail := func(format string, args ...any) error { - return harness.BailInstance(env["RESULTS_FILE"], "Backfill ingestion", env["RUN_ID"], env["TARGET_SHA"], + return harness.BailInstance(cfg.ResultsFile, "Backfill ingestion", cfg.RunID, cfg.TargetSHA, fmt.Sprintf(format, args...)) } - want, err := strconv.Atoi(retention) // used to compare against ingested ledgers below - if err != nil { - return bail("parsing retention window %q: %v", retention, err) - } - - awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(env["REGION"])) + awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(cfg.Region)) if err != nil { return bail("loading AWS config: %v", err) } - fetch := &harness.S3Fetcher{Client: s3.NewFromConfig(awsCfg), Bucket: env["BUCKET"]} + fetch := &harness.S3Fetcher{Client: s3.NewFromConfig(awsCfg), Bucket: cfg.Bucket} - if err := fetch.FetchVerified(ctx, "core/stellar-core.zst", corePath, true, "stellar-core"); err != nil { + coreCfg, err := prepareFixtures(ctx, fetch, repoRoot, cfg.WorkDir, binaryPath) + if err != nil { return bail("%v", err) } - if err := os.Chmod(corePath, 0o755); err != nil { - return bail("chmod stellar-core: %v", err) - } - - logger.Infof("building stellar-rpc") - if err := harness.RunStreaming(ctx, repoRoot, nil, 40, "make", "build-libs"); err != nil { - return bail("make build-libs failed: %v", err) - } - if err := harness.RunStreaming(ctx, repoRoot, nil, 40, - "go", "build", "-o", binaryPath, "./cmd/stellar-rpc"); err != nil { - return bail("go build failed: %v", err) - } - // fetch + write core config from SDK - coreCfg := filepath.Join(env["WORK_DIR"], "captive-core-pubnet.cfg") - if err := os.WriteFile(coreCfg, ledgerbackend.PubnetDefaultConfig, 0o644); err != nil { - return bail("writing captive-core config: %v", err) + endpoint := "localhost:" + rpcPort + if cfg.ServeAfter { + endpoint = "0.0.0.0:" + rpcPort } - cfgPath, err := renderConfig(repoRoot, env["WORK_DIR"], coreCfg, retention) + cfgPath, err := renderConfig(repoRoot, cfg.WorkDir, coreCfg, retention, endpoint) if err != nil { return bail("rendering config: %v", err) } - dl, err := time.ParseDuration(deadline) - if err != nil { - return bail("invalid BACKFILL_DEADLINE %q: %v", deadline, err) - } - logger.Infof("starting backfill (retention=%s, deadline=%s)", retention, deadline) - elapsed, lo, hi, err := runBackfill(ctx, dl, binaryPath, cfgPath) + logger.Infof("starting backfill (retention=%s, deadline=%s, serve-after=%t)", retention, cfg.Deadline, cfg.ServeAfter) + elapsed, lo, hi, daemon, err := runBackfill(ctx, cfg.Deadline, binaryPath, cfgPath, cfg.ServeAfter) if err != nil { return bail("%v", err) } + if daemon != nil { + defer daemon.Stop() // covers the bail paths below; Stop is idempotent + } ingested := hi - lo + 1 - if ingested+ledgerThreshold < want { + if ingested+ledgerThreshold < cfg.Retention { return bail("backfill reported complete but ingested %d of %s ledgers", ingested, retention) } logger.Infof("backfill complete: %d ledgers [%d -> %d] in %s", ingested, lo, hi, elapsed.Round(time.Second)) - md := renderMarkdown(env["TARGET_SHA"], retention, lo, hi, ingested, elapsed) - if err := os.WriteFile(env["RESULTS_FILE"], []byte(md), 0o644); err != nil { + md := renderMarkdown(cfg.TargetSHA, retention, lo, hi, ingested, elapsed) + if err := os.WriteFile(cfg.ResultsFile, []byte(md), 0o644); err != nil { return bail("writing results: %v", err) } if err := harness.PublishResult( - ctx, fetch.Client, env["BUCKET"], env["RESULT_KEY"], "ok", env["RUN_ID"], env["TARGET_SHA"], - env["RESULTS_FILE"], ""); err != nil { + ctx, fetch.Client, cfg.Bucket, cfg.ResultKey, "ok", cfg.RunID, cfg.TargetSHA, cfg.ResultsFile, ""); err != nil { return bail("publishing result: %v", err) } + + if daemon != nil { + // hand the serving box off to the chained blaster leg + servePhase(ctx, daemon) + } return nil } -// renderConfig fills the config template's ${...} placeholders (box paths + the -// retention window) via os.Expand -func renderConfig(repoRoot, workDir, coreCfg, retention string) (string, error) { +// prepareFixtures fetches stellar-core, builds stellar-rpc into binaryPath, +// and writes the SDK's captive-core pubnet config, returning its path. +func prepareFixtures( + ctx context.Context, fetch *harness.S3Fetcher, repoRoot, workDir, binaryPath string, +) (string, error) { + if err := fetch.FetchVerified(ctx, "core/stellar-core.zst", corePath, true, "stellar-core"); err != nil { + return "", err + } + if err := os.Chmod(corePath, 0o755); err != nil { + return "", fmt.Errorf("chmod stellar-core: %w", err) + } + + logger.Infof("building stellar-rpc") + if err := harness.RunStreaming(ctx, repoRoot, nil, 40, "make", "build-libs"); err != nil { + return "", fmt.Errorf("make build-libs failed: %w", err) + } + if err := harness.RunStreaming(ctx, repoRoot, nil, 40, + "go", "build", "-o", binaryPath, "./cmd/stellar-rpc"); err != nil { + return "", fmt.Errorf("go build failed: %w", err) + } + + // fetch + write core config from SDK + coreCfg := filepath.Join(workDir, "captive-core-pubnet.cfg") + if err := os.WriteFile(coreCfg, ledgerbackend.PubnetDefaultConfig, 0o644); err != nil { + return "", fmt.Errorf("writing captive-core config: %w", err) + } + return coreCfg, nil +} + +// renderConfig fills the config template's ${...} placeholders (box paths, the +// retention window, and the bind endpoint) via os.Expand +func renderConfig(repoRoot, workDir, coreCfg, retention, endpoint string) (string, error) { tmpl, err := os.ReadFile(filepath.Join(repoRoot, legDir, "testdata", "backfill-pubnet.toml.tmpl")) if err != nil { return "", err @@ -134,6 +165,8 @@ func renderConfig(repoRoot, workDir, coreCfg, retention string) (string, error) return corePath case "HISTORY_RETENTION_WINDOW": return retention + case "ENDPOINT": + return endpoint default: return "${" + in + "}" // leave unknown placeholders intact } @@ -146,18 +179,38 @@ func renderConfig(repoRoot, workDir, coreCfg, retention string) (string, error) return cfgPath, nil } +// daemonHandle controls a daemon left running past its backfill phase. +type daemonHandle struct { + cancel context.CancelFunc + done chan struct{} // closed once the daemon is reaped +} + +// Stop kills the daemon and waits (bounded) for it to be reaped. +func (d *daemonHandle) Stop() { + d.cancel() + select { + case <-d.done: + case <-time.After(30 * time.Second): + logger.Warnf("daemon not reaped within 30s of stop") + } +} + // runBackfill launches the daemon and streams its output (teeing to the box log) -// until the backfill-complete line fires, recording the wall-clock -func runBackfill(ctx context.Context, deadline time.Duration, binary, cfgPath string) (time.Duration, int, int, error) { - runCtx, cancel := context.WithTimeout(ctx, deadline) - defer cancel() +// until the backfill-complete line fires, recording the wall-clock. +func runBackfill( + ctx context.Context, deadline time.Duration, binary, cfgPath string, keepAlive bool, +) (time.Duration, int, int, *daemonHandle, error) { + runCtx, cancel := context.WithCancel(ctx) + // deadline covers only the backfill phase, which the daemon may outlive + watchdog := time.AfterFunc(deadline, cancel) cmd := exec.CommandContext(runCtx, binary, "--config-path", cfgPath) // hide this box's IMDS creds as the public datalake 403s signed requests cmd.Env = append(os.Environ(), "AWS_EC2_METADATA_DISABLED=true") pr, pw, err := os.Pipe() if err != nil { - return 0, 0, 0, err + cancel() + return 0, 0, 0, nil, err } cmd.Stdout, cmd.Stderr = pw, pw @@ -165,10 +218,10 @@ func runBackfill(ctx context.Context, deadline time.Duration, binary, cfgPath st if err := cmd.Start(); err != nil { pw.Close() pr.Close() - return 0, 0, 0, fmt.Errorf("starting daemon: %w", err) + cancel() + return 0, 0, 0, nil, fmt.Errorf("starting daemon: %w", err) } pw.Close() // the child holds the write end and we read until it dies - defer pr.Close() var elapsed time.Duration var lo, hi int @@ -181,23 +234,38 @@ func runBackfill(ctx context.Context, deadline time.Duration, binary, cfgPath st elapsed = time.Since(start) lo, _ = strconv.Atoi(m[1]) hi, _ = strconv.Atoi(m[2]) - cancel() // stop the daemon before it starts live ingestion + watchdog.Stop() break } } - scanErr := scanner.Err() - if scanErr != nil { + + if elapsed == 0 { // daemon died, read failure, or the watchdog killed it cancel() + pr.Close() + _ = cmd.Wait() + if scanErr := scanner.Err(); scanErr != nil { + return 0, 0, 0, nil, fmt.Errorf("reading daemon output: %w", scanErr) + } + return 0, 0, 0, nil, fmt.Errorf("daemon exited or hit the %s deadline before backfill completed", deadline) } - _ = cmd.Wait() // reap; a kill from cancel surfaces here and is expected - if elapsed == 0 { - if scanErr != nil { - return 0, 0, 0, fmt.Errorf("reading daemon output: %w", scanErr) + // keep draining the pipe (the daemon blocks on it once full) and teeing + // its catchup/ingestion output to the box log until it dies + done := make(chan struct{}) + go func() { + defer close(done) + for scanner.Scan() { + fmt.Fprintln(os.Stderr, scanner.Text()) } - return 0, 0, 0, fmt.Errorf("daemon exited or hit the %s deadline before backfill completed", deadline) + pr.Close() + _ = cmd.Wait() // reap; a kill from cancel surfaces here and is expected + }() + daemon := &daemonHandle{cancel: cancel, done: done} + if !keepAlive { + daemon.Stop() // stop the daemon before it starts live ingestion + return elapsed, lo, hi, nil, nil } - return elapsed, lo, hi, nil + return elapsed, lo, hi, daemon, nil } func renderMarkdown(sha, retention string, lo, hi, ingested int, elapsed time.Duration) string { diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/main.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/main.go index 5f2acfa2a..371fcce51 100644 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/main.go +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/main.go @@ -1,4 +1,5 @@ -// Command runner runs the backfill ingestion benchmark on the box. +// Command runner runs the backfill ingestion benchmark on the box (then keeps +// serving in handoff mode). package main import "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness" diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/serve.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/serve.go new file mode 100644 index 000000000..fb8fd9c29 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/serve.go @@ -0,0 +1,75 @@ +package main + +import ( + "context" + "fmt" + "os/exec" + "path" + "time" + + "github.com/caarlos0/env/v11" +) + +const rpcPort = "8000" // rpcPort is where the daemon serves, crosses VPC + +// serveEnv is the serve phase's env-derived config. +type serveEnv struct { + Ceiling time.Duration `env:"SERVE_CEILING" envDefault:"6h"` + Bucket string `env:"BUCKET"` + ResultKey string `env:"RESULT_KEY"` +} + +// servePhase parks the box serving for the blaster leg, which gets this box's +// IP through the coordinator. Termination is through other job's adopt cleanup. +func servePhase(ctx context.Context, daemon *daemonHandle) { + // on the backstop path the leg script's EXIT trap gets a minute to upload + // the box log before poweroff + defer rescheduleShutdown(ctx, 1, "serve phase over") + defer daemon.Stop() + + cfg, err := env.ParseAs[serveEnv]() + if err != nil { + logger.Warnf("parsing serve env: %v", err) + cfg.Ceiling = 6 * time.Hour + } + + // the boot-time self-terminate (budget+15m) can predate the blast's end + rescheduleShutdown(ctx, int(cfg.Ceiling.Minutes()), "serve ceiling") + + // the happy-path hard kill skips the EXIT-trap upload, persist the log up to now + snapshotBoxLog(ctx, cfg.Bucket, cfg.ResultKey) + + logger.Infof("serving :%s until external termination (ceiling %s)", rpcPort, cfg.Ceiling) + select { + case <-ctx.Done(): + case <-time.After(cfg.Ceiling): + logger.Warnf("serve ceiling passed without external termination") // backstops orphaned runs + } +} + +// snapshotBoxLog copies the box log so far next to the result object. +func snapshotBoxLog(ctx context.Context, bucket, key string) { + if bucket == "" || key == "" { + return + } + dst := fmt.Sprintf("s3://%s/%s/user-data-serving.log", bucket, path.Dir(key)) + cmd := exec.CommandContext(ctx, "aws", "s3", "cp", "/var/log/user-data.log", dst) + if out, err := cmd.CombinedOutput(); err != nil { + logger.Warnf("snapshotting box log to %s: %v (%s)", dst, err, out) + } +} + +// rescheduleShutdown replaces the box's pending `shutdown -P` with one minutes +// from now (shutdown behavior is terminate). Best-effort: a failure only means +// the box may power off off-schedule, never that it leaks. +func rescheduleShutdown(ctx context.Context, minutes int, reason string) { + if out, err := exec.CommandContext(ctx, "shutdown", "-c").CombinedOutput(); err != nil { + logger.Warnf("canceling pending shutdown: %v (%s)", err, out) + } + arg := fmt.Sprintf("+%d", minutes) + if out, err := exec.CommandContext(ctx, "shutdown", "-P", arg, reason).CombinedOutput(); err != nil { + logger.Warnf("rescheduling shutdown: %v (%s)", err, out) + } else { + logger.Infof("shutdown rescheduled to %s from now (%s)", arg, reason) + } +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/testdata/backfill-pubnet.toml.tmpl b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/testdata/backfill-pubnet.toml.tmpl index 80927c9fc..7cb11d733 100644 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/testdata/backfill-pubnet.toml.tmpl +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/testdata/backfill-pubnet.toml.tmpl @@ -4,7 +4,7 @@ NETWORK_PASSPHRASE = "Public Global Stellar Network ; September 2015" HISTORY_ARCHIVE_URLS = ["https://history.stellar.org/prd/core-live/core_live_001"] -ENDPOINT = "localhost:8000" +ENDPOINT = "${ENDPOINT}" LOG_LEVEL = "info" # Captive core is constructed at startup (so the binary + config must exist) diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/run-blaster.sh b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/run-blaster.sh new file mode 100644 index 000000000..2fe272b3e --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/run-blaster.sh @@ -0,0 +1,7 @@ +# Endpoint load-test leg (chained after the backfill leg). Concatenated after +# bootstrap-common.sh in the EC2 user-data, so it relies on that file's env, +# helpers, bootstrap_box, and run_leg. The runner blasts the backfill box's RPC. +LEG_TITLE="Endpoint load test" + +bootstrap_box +run_leg ./cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/instantiate.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/instantiate.go new file mode 100644 index 000000000..ca0751cfd --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/instantiate.go @@ -0,0 +1,180 @@ +package main + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + "github.com/caarlos0/env/v11" + + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness" +) + +// blasterCfg is the endpoint roster, shipped with the blaster checkout. +const blasterCfg = "cmd/stellar-rpc-blaster/internal/config/config.example.toml" + +// blasterEnv is the leg's env-derived config. +type blasterEnv struct { + RampUp string `env:"BLASTER_RAMP_UP" envDefault:"2m"` + Duration string `env:"BLASTER_DURATION" envDefault:"3m"` + // recovery gap between serial endpoints, so one endpoint's failures + // don't cascade into the next + Cooloff string `env:"BLASTER_COOLOFF" envDefault:"15s"` + SeedCount string `env:"SEED_COUNT" envDefault:"1000"` + // left buffer outruns retention trimming during the blast; right buffer + // keeps clear of the (still advancing) tip + BufferLow int64 `env:"SEED_BUFFER_LOW" envDefault:"1000"` + BufferHigh int64 `env:"SEED_BUFFER_HIGH" envDefault:"128"` + // serving box's address, passed by the coordinator once the backfill leg passes + TargetRPC string `env:"TARGET_RPC"` + CatchupTimeout time.Duration `env:"CATCHUP_TIMEOUT" envDefault:"60m"` + BudgetMinutes int `env:"BUDGET_MINUTES"` + BlasterRepo string `env:"BLASTER_REPO" envDefault:"stellar/stellar-rpc-blaster"` +} + +// instantiate is the instance's blast task: it receives the chained peer's serving +// RPC, generates seed data, runs the endpoint blast, and publishes the stats. +func instantiate(ctx context.Context) error { + leg, err := harness.LegSetup(ctx, "Endpoint load test") + if err != nil { + return err + } + cfg, err := env.ParseAs[blasterEnv]() + if err != nil { + return leg.Bail("parsing env: %v", err) + } + + if deadline, ok := harness.BootDeadline(cfg.BudgetMinutes, 25*time.Minute); ok { + var cancel context.CancelFunc + ctx, cancel = context.WithDeadline(ctx, deadline) + defer cancel() + logger.Infof("leg deadline in %s (budget-derived)", time.Until(deadline).Round(time.Minute)) + } + + if cfg.TargetRPC == "" { + return leg.Bail("TARGET_RPC unset; nothing to blast") + } + + // fetch + build overlap the target box's catchup + blasterDir := filepath.Join(leg.WorkDir, "stellar-rpc-blaster") + blasterBin, blasterSHA, err := fetchBlaster(ctx, blasterDir, cfg.BlasterRepo) + if err != nil { + return leg.Bail("%v", err) + } + + wctx, wcancel := context.WithTimeout(ctx, cfg.CatchupTimeout) + waitStart := time.Now() + health, err := harness.AwaitHealthy(wctx, cfg.TargetRPC, 15*time.Second) // await catchup + wcancel() + if err != nil { + return leg.Bail("target RPC %s: %v", cfg.TargetRPC, err) + } + handoffSecs := int(time.Since(waitStart).Seconds()) + logger.Infof("target RPC %s serving ledgers [%d, %d] (handoff wait %ds)", + cfg.TargetRPC, health.OldestLedger, health.LatestLedger, handoffSecs) + + lo, hi := int64(health.OldestLedger)+cfg.BufferLow, int64(health.LatestLedger)-cfg.BufferHigh + if hi <= lo { + return leg.Bail("ledger window [%d, %d] leaves no room after buffers +%d/-%d", + health.OldestLedger, health.LatestLedger, cfg.BufferLow, cfg.BufferHigh) + } + + // launch blast + call := blastCall{ + bin: blasterBin, url: cfg.TargetRPC, + configPath: filepath.Join(blasterDir, blasterCfg), + seedPath: filepath.Join(leg.WorkDir, "blaster-seed.json"), + resultsPath: filepath.Join(leg.WorkDir, "blaster-results.json"), + rampUp: cfg.RampUp, duration: cfg.Duration, cooloff: cfg.Cooloff, + } + if err := generateSeed(ctx, call, lo, hi, cfg.SeedCount); err != nil { + return leg.Bail("%v", err) + } + if err := blast(ctx, call); err != nil { + return leg.Bail("%v", err) + } + data, err := os.ReadFile(call.resultsPath) + if err != nil { + return leg.Bail("reading blaster results: %v", err) + } + rows, err := summarize(data) + if err != nil { + return leg.Bail("summarizing blaster results: %v", err) + } + + md := renderMarkdown(leg.TargetSHA, blasterSHA, cfg.RampUp, cfg.Duration, + health.OldestLedger, health.LatestLedger, handoffSecs, rows) + if err := os.WriteFile(leg.ResultsFile, []byte(md), 0o644); err != nil { + return leg.Bail("writing results: %v", err) + } + if err := leg.Publish(ctx, call.resultsPath); err != nil { + return leg.Bail("publishing result: %v", err) + } + return nil +} + +// fetchBlaster clones and builds stellar-rpc-blaster at dev HEAD +func fetchBlaster(ctx context.Context, dir, repo string) (string, string, error) { + logger.Infof("fetching stellar-rpc-blaster (%s@dev)", repo) + if err := os.RemoveAll(dir); err != nil { + return "", "", err + } + if err := harness.RunStreaming(ctx, "", nil, 20, "git", "clone", "-q", "--depth", "1", + "--branch", "dev", "https://github.com/"+repo+".git", dir); err != nil { + return "", "", fmt.Errorf("git clone failed: %w", err) + } + out, err := exec.CommandContext(ctx, "git", "-C", dir, "rev-parse", "HEAD").Output() + if err != nil { + return "", "", fmt.Errorf("resolving blaster commit: %w", err) + } + sha := strings.TrimSpace(string(out)) + + logger.Infof("building stellar-rpc-blaster at %s", sha) + if err := harness.RunStreaming(ctx, dir, nil, 40, "make", "build"); err != nil { + return "", "", fmt.Errorf("blaster build failed: %w", err) + } + return filepath.Join(dir, "stellar-rpc-blaster"), sha, nil +} + +// blastCall parameterizes one serial blaster sweep. +type blastCall struct { + bin, url string + configPath string + seedPath, resultsPath string + rampUp, duration, cooloff string +} + +// generateSeed samples the request corpus from the target RPC's ledger window. +func generateSeed(ctx context.Context, c blastCall, lo, hi int64, count string) error { + logger.Infof("generating seed data: %s ledgers sampled from [%d, %d]", count, lo, hi) + if err := harness.RunStreaming(ctx, filepath.Dir(c.bin), nil, 40, c.bin, "generate", + "--rpc-url", c.url, + "--output", c.seedPath, + "--ledger-window", fmt.Sprintf("%d,%d", lo, hi), + "--count", count); err != nil { + return fmt.Errorf("blaster generate failed: %w", err) + } + return nil +} + +// blast runs the serial endpoint sweep, writing results to c.resultsPath. +func blast(ctx context.Context, c blastCall) error { + logger.Infof("blasting endpoints in serial (ramp-up %s, duration %s, cooloff %s per endpoint)", + c.rampUp, c.duration, c.cooloff) + if err := harness.RunStreaming(ctx, filepath.Dir(c.bin), nil, 80, c.bin, "run", + "--rpc-url", c.url, + "--config-path", c.configPath, + "--input-data-path", c.seedPath, + "--serial", + "--ramp-up", c.rampUp, + "--duration", c.duration, + "--cooloff", c.cooloff, + "--test-output-path", c.resultsPath); err != nil { + return fmt.Errorf("blaster run failed: %w", err) + } + return nil +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/main.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/main.go new file mode 100644 index 000000000..ee9415f80 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/main.go @@ -0,0 +1,10 @@ +// Command runner drives the endpoint load-test leg's on-box half: it waits for +// the backfill box's serve-ready object, seeds request data from it, and blasts +// its read-path endpoints. The GHA-runner half is the shared perf-eval/gather. +package main + +import "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness" + +var logger = harness.NewLogger() + +func main() { harness.Run(instantiate) } diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/results.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/results.go new file mode 100644 index 000000000..db7da42fa --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/results.go @@ -0,0 +1,85 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" +) + +// endpointStats is one endpoint's row of the report, distilled from blaster's +// results JSON (percentile keys there are "p50.0", "p95.0", "p99.0", "p99.9"). +type endpointStats struct { + Name string + TargetRPS float64 + Limit uint64 // effective pagination limit; 0 = endpoint doesn't paginate + Requests uint64 + Errors uint64 + P50 float64 + P95 float64 + P99 float64 + P999 float64 +} + +// summarize turns blaster's results JSON into per-endpoint rows. +func summarize(data []byte) ([]endpointStats, error) { + var res struct { + //nolint:tagliatelle // external schema: blaster emits snake_case + Endpoints map[string]struct { + TotalRequests uint64 `json:"total_requests"` + Errors uint64 `json:"errors"` + TargetRPS float64 `json:"target_rps"` + Limit uint64 `json:"limit"` + Percentiles map[string]float64 `json:"percentiles_ms"` + } `json:"endpoints"` + } + if err := json.Unmarshal(data, &res); err != nil { + return nil, err + } + if len(res.Endpoints) == 0 { + return nil, errors.New("blaster results hold no endpoints") + } + + rows := make([]endpointStats, 0, len(res.Endpoints)) + for name, ep := range res.Endpoints { + rows = append(rows, endpointStats{ + Name: name, + TargetRPS: ep.TargetRPS, + Limit: ep.Limit, + Requests: ep.TotalRequests, + Errors: ep.Errors, + P50: ep.Percentiles["p50.0"], + P95: ep.Percentiles["p95.0"], + P99: ep.Percentiles["p99.0"], + P999: ep.Percentiles["p99.9"], + }) + } + sort.Slice(rows, func(i, j int) bool { return rows[i].Name < rows[j].Name }) + return rows, nil +} + +func renderMarkdown( + sha, blasterSHA, rampUp, duration string, oldest, latest uint32, handoffSecs int, rows []endpointStats, +) string { + var b strings.Builder + fmt.Fprintf(&b, "### 🎯 Endpoint load test — `%s`\n\n", sha[:min(12, len(sha))]) + fmt.Fprintf(&b, "Serial blast per endpoint (ramp-up %s, duration %s, blaster `%s`) against the backfilled RPC "+ + "(ledgers `[%d, %d]`, handoff wait %ds).\n\n", + rampUp, duration, blasterSHA[:min(12, len(blasterSHA))], oldest, latest, handoffSecs) + b.WriteString("| Endpoint | Target RPS | Requests | Errors | p50 (ms) | p95 (ms) | p99 (ms) | p99.9 (ms) |\n") + b.WriteString("|---|---|---|---|---|---|---|---|\n") + for _, r := range rows { + errPct := 0.0 + if r.Requests > 0 { + errPct = float64(r.Errors) / float64(r.Requests) * 100 + } + name := r.Name + if r.Limit > 0 { + name = fmt.Sprintf("%s (limit=%d)", r.Name, r.Limit) + } + fmt.Fprintf(&b, "| %s | %.0f | %d | %d (%.1f%%) | %.1f | %.1f | %.1f | %.1f |\n", + name, r.TargetRPS, r.Requests, r.Errors, errPct, r.P50, r.P95, r.P99, r.P999) + } + return b.String() +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/results_test.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/results_test.go new file mode 100644 index 000000000..93ef30fdc --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/endpoint-load-test/runner/results_test.go @@ -0,0 +1,76 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// fixture mirrors blaster's results JSON shape, including the fields the +// summary drops (timeline, error_types, top-level timings). +const fixture = `{ + "start": "2026-07-10T00:00:00Z", + "end": "2026-07-10T00:06:10Z", + "duration_seconds": 370, + "endpoints": { + "getLedgers": { + "total_requests": 3600, + "success": 3591, + "errors": 9, + "target_rps": 20, + "limit": 1, + "percentiles_ms": {"p50.0": 3.2, "p95.0": 9.8, "p99.0": 21.5, "p99.9": 60.1}, + "error_types": {"rpc_error": {"error_msg": "boom", "error_code": -32600, "count": 9}}, + "timeline": [ + {"target_rps": 2, "success": 10, "errors": 0, "error_rate_pct": 0, + "p50_ms": 3, "p95_ms": 9, "p99_ms": 20, "p99.9_ms": 55} + ] + }, + "getHealth": { + "total_requests": 18000, + "success": 18000, + "errors": 0, + "target_rps": 100, + "percentiles_ms": {"p50.0": 0.4, "p95.0": 0.9, "p99.0": 1.5, "p99.9": 4.2} + } + } +}` + +func TestSummarize(t *testing.T) { + rows, err := summarize([]byte(fixture)) + require.NoError(t, err) + require.Len(t, rows, 2) + + // sorted by endpoint name + require.Equal(t, "getHealth", rows[0].Name) + require.Equal(t, "getLedgers", rows[1].Name) + + gl := rows[1] + require.Equal(t, uint64(3600), gl.Requests) + require.Equal(t, uint64(9), gl.Errors) + require.Equal(t, uint64(1), gl.Limit) + require.Zero(t, rows[0].Limit) // getHealth doesn't paginate + require.InDelta(t, 20.0, gl.TargetRPS, 0.001) + require.InDelta(t, 3.2, gl.P50, 0.001) + require.InDelta(t, 9.8, gl.P95, 0.001) + require.InDelta(t, 21.5, gl.P99, 0.001) + require.InDelta(t, 60.1, gl.P999, 0.001) +} + +func TestRenderMarkdown(t *testing.T) { + // fails on empty + _, err := summarize([]byte(`{"endpoints": {}}`)) + require.Error(t, err) + + rows, err := summarize([]byte(fixture)) + require.NoError(t, err) + md := renderMarkdown("0123456789abcdef", "fedcba9876543210", "2m", "3m", 60_000_000, 60_017_280, 1800, rows) + + require.Contains(t, md, "`0123456789ab`") + require.Contains(t, md, "ramp-up 2m, duration 3m, blaster `fedcba987654`") + require.Contains(t, md, "`[60000000, 60017280]`") + require.Contains(t, md, "handoff wait 1800s") + require.Contains(t, md, "| p50 (ms) | p95 (ms) | p99 (ms) | p99.9 (ms) |") + require.Contains(t, md, "| getLedgers (limit=1) | 20 | 3600 | 9 (0.2%) | 3.2 | 9.8 | 21.5 | 60.1 |") + require.Contains(t, md, "| getHealth | 100 | 18000 | 0 (0.0%) | 0.4 | 0.9 | 1.5 | 4.2 |") +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness.go index 134848982..4cd3ebe9c 100644 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness.go +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness.go @@ -20,6 +20,7 @@ import ( "fmt" "os" "strings" + "time" supportlog "github.com/stellar/go-stellar-sdk/support/log" ) @@ -64,6 +65,39 @@ func Env(key, def string) string { return def } +// RequireEnv returns the values of keys in order, erroring with every unset one. +func RequireEnv(keys ...string) ([]string, error) { + vals := make([]string, len(keys)) + var missing []string + for i, k := range keys { + if vals[i] = os.Getenv(k); vals[i] == "" { + missing = append(missing, k) + } + } + if len(missing) > 0 { + return nil, fmt.Errorf("missing required env: %s", strings.Join(missing, ", ")) + } + return vals, nil +} + +// BootDeadline returns the instant a box-side runner should bail by: budget +// minutes after box boot, minus margin. ok is false when the budget is unset. +func BootDeadline(budgetMinutes int, margin time.Duration) (time.Time, bool) { + if budgetMinutes <= 0 { + return time.Time{}, false + } + up, err := os.ReadFile("/proc/uptime") + if err != nil { + return time.Time{}, false + } + var uptimeSecs float64 + if _, err := fmt.Sscanf(string(up), "%f", &uptimeSecs); err != nil { + return time.Time{}, false + } + boot := time.Now().Add(-time.Duration(uptimeSecs * float64(time.Second))) + return boot.Add(time.Duration(budgetMinutes)*time.Minute - margin), true +} + // appendOutputs appends lines to the GitHub Actions step-output file. func appendOutputs(path string, lines ...string) error { f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0o644) diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/health.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/health.go new file mode 100644 index 000000000..9459c46c1 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/health.go @@ -0,0 +1,35 @@ +package harness + +import ( + "context" + "fmt" + "time" + + "github.com/stellar/go-stellar-sdk/clients/rpcclient" + protocol "github.com/stellar/go-stellar-sdk/protocols/rpc" +) + +// AwaitHealthy polls url's getHealth every interval until the RPC reports +// healthy, returning the winning response. ctx bounds the wait. +func AwaitHealthy(ctx context.Context, url string, interval time.Duration) (protocol.GetHealthResponse, error) { + client := rpcclient.NewClient(url, nil) + defer client.Close() + for poll := 0; ; poll++ { + res, err := client.GetHealth(ctx) + if err == nil && res.Status == "healthy" { + return res, nil + } + last := fmt.Sprintf("status %q", res.Status) + if err != nil { + last = err.Error() + } + if poll%4 == 0 { + logger.Infof("waiting for %s to report healthy (last: %s)", url, last) + } + select { + case <-ctx.Done(): + return protocol.GetHealthResponse{}, fmt.Errorf("%s never reported healthy (last: %s): %w", url, last, ctx.Err()) + case <-time.After(interval): + } + } +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/leg.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/leg.go new file mode 100644 index 000000000..1bc6ce280 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/leg.go @@ -0,0 +1,57 @@ +package harness + +import ( + "context" + "fmt" + "os" + + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/caarlos0/env/v11" +) + +// Leg is the env-derived context shared by every on-box leg runner. +type Leg struct { + Title string + Bucket string `env:"BUCKET" envDefault:"stellar-rpc-ci-load-test"` + Region string `env:"REGION" envDefault:"us-east-1"` + WorkDir string `env:"WORK_DIR" envDefault:"/data"` + ResultsFile string `env:"RESULTS_FILE" envDefault:"/tmp/results.md"` + ResultKey string `env:"RESULT_KEY"` + TargetSHA string `env:"TARGET_SHA"` + RunID string `env:"RUN_ID" envDefault:"manual"` + RepoRoot string // cwd: run_leg starts runners at the repo checkout root + Fetch *S3Fetcher +} + +// LegSetup reads the common leg env and builds the S3 client. On failure it +// has already written the bail body, so the runner just returns the error. +func LegSetup(ctx context.Context, title string) (*Leg, error) { + l := &Leg{Title: title} + if err := env.Parse(l); err != nil { + return nil, fmt.Errorf("parsing leg env: %w", err) + } + var err error + if l.RepoRoot, err = os.Getwd(); err != nil { + return nil, l.Bail("getwd: %v", err) + } + awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(l.Region)) + if err != nil { + return nil, l.Bail("loading AWS config: %v", err) + } + l.Fetch = &S3Fetcher{Client: s3.NewFromConfig(awsCfg), Bucket: l.Bucket} + return l, nil +} + +// Bail writes the leg's failure body and returns the error for the runner to +// exit with (run_leg then publishes the fail verdict). +func (l *Leg) Bail(format string, args ...any) error { + return BailInstance(l.ResultsFile, l.Title, l.RunID, l.TargetSHA, fmt.Sprintf(format, args...)) +} + +// Publish uploads the ok verdict with the leg's results markdown (and bench +// JSON when benchPath is non-empty). +func (l *Leg) Publish(ctx context.Context, benchPath string) error { + return PublishResult(ctx, l.Fetch.Client, l.Bucket, l.ResultKey, "ok", + l.RunID, l.TargetSHA, l.ResultsFile, benchPath) +} diff --git a/go.mod b/go.mod index d2a250e3e..7c9e56e26 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/s3 v1.89.1 github.com/aws/aws-sdk-go-v2/service/ssm v1.69.3 github.com/aws/smithy-go v1.27.1 + github.com/caarlos0/env/v11 v11.4.1 github.com/cenkalti/backoff/v4 v4.3.0 github.com/creachadair/jrpc2 v1.3.3 github.com/fsouza/fake-gcs-server v1.49.2 diff --git a/go.sum b/go.sum index 9a94100f5..964eec7c3 100644 --- a/go.sum +++ b/go.sum @@ -130,6 +130,8 @@ github.com/aws/smithy-go v1.27.1 h1:4T340VFndXtADGF52gYa1POyL7s9E4Z1OeZ1hCscIw8= github.com/aws/smithy-go v1.27.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/caarlos0/env/v11 v11.4.1 h1:fYwH0sWEsBSMPG7t4e/PEfTFzrWrpjyygXyUnWiSwEw= +github.com/caarlos0/env/v11 v11.4.1/go.mod h1:qupehSf/Y0TUTsxKywqRt/vJjN5nz6vauiYEUUr8P4U= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=