diff --git a/.github/workflows/ec2-leg.yml b/.github/workflows/ec2-leg.yml new file mode 100644 index 000000000..f3cbb21d7 --- /dev/null +++ b/.github/workflows/ec2-leg.yml @@ -0,0 +1,264 @@ +name: Perf-eval EC2 leg (ephemeral) +# Generic callable leg: launches an ephemeral EC2 box, renders its user-data from +# bootstrap-common.sh + the leg's own run script, polls S3 for the result object +# the box publishes, and reports through that object (job result = leg verdict). +# 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. + +on: + # workflow_call is the only trigger: a leg runs through the coordinator. + workflow_call: + inputs: + target_ref: + description: 'Ref or SHA to benchmark (resolved to a SHA on checkout)' + required: true + type: string + run_label: + description: 'Leg label; folded into the S3 result key and instance name' + required: true + type: string + leg_script: + description: 'Path to the leg run script, concatenated after bootstrap-common.sh into the box user-data' + required: true + type: string + instance_type: + description: 'EC2 instance type' + required: false + type: string + default: c5.2xlarge + root_volume_gb: + description: 'Root gp3 volume size (GB)' + required: false + type: number + default: 500 + volume_iops: + description: 'Root gp3 IOPS (3000 is the gp3 floor)' + required: false + type: number + default: 3000 + volume_throughput: + description: 'Root gp3 throughput (MiB/s)' + required: false + type: number + default: 125 + ssm_registration_timeout: + description: 'Seconds to wait for the SSM agent to register' + required: false + type: number + default: 240 + budget_minutes: + description: 'End-to-end leg ceiling (= job timeout); S3 poll deadline, box self-terminate, and OIDC session derive from it (-15m/+15m/+15m). Derived session must fit the IAM role max.' + required: false + type: number + default: 225 + poll_interval: + description: 'Seconds between S3 result polls' + required: false + type: number + default: 30 + extra_env: + description: 'Extra `export FOO=bar` lines injected into the box user-data preamble' + required: false + type: string + default: '' + secrets: + AWS_GHA_ROLE_ARN: + description: 'OIDC role ARN to assume for EC2 + S3 access' + required: true + +# Only the intrinsic infra permissions necessary +permissions: + id-token: write # OIDC AssumeRole into the GHA role + contents: read # checkout + +jobs: + ec2-leg: + name: Launch + await ephemeral perf-eval box + # 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: + group: ec2-leg-${{ inputs.target_ref }}-${{ inputs.run_label }} + cancel-in-progress: false + runs-on: ubuntu-latest + timeout-minutes: ${{ inputs.budget_minutes }} + env: + AWS_REGION: us-east-1 + INSTANCE_TYPE: ${{ inputs.instance_type }} + ROOT_VOLUME_GB: ${{ inputs.root_volume_gb }} + BOOTSTRAP_VOLUME_IOPS: ${{ inputs.volume_iops }} + BOOTSTRAP_VOLUME_THROUGHPUT: ${{ inputs.volume_throughput }} + INSTANCE_PROFILE: stellar-rpc-ci-load-test + TEST_TAG_KEY: test + TEST_TAG_VAL: stellar-rpc-ci-load-test + SSM_REGISTRATION_TIMEOUT: ${{ inputs.ssm_registration_timeout }} + POLL_INTERVAL: ${{ inputs.poll_interval }} + DEBUG_LOG_LINES: 40 + DEBUG_LOG_EVERY_POLLS: 5 + PERF_EVAL_DIR: cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval + LEG_SCRIPT: ${{ inputs.leg_script }} + BUCKET: stellar-rpc-ci-load-test + RESULT_KEY: runs/${{ github.run_id }}/${{ inputs.run_label }}/result.json # deterministic per run + leg + + steps: + # budget_minutes is the job timeout and the other deadlines derive from it so + # the ordering (poll < job < self-terminate <= role session) holds by + # construction. + - name: Derive deadlines + id: deadlines + run: | + B=${{ inputs.budget_minutes }} + [ "$B" -gt 30 ] || { echo "::error::budget_minutes must exceed the 15m infra buffers"; exit 1; } + { + echo "RESULTS_TIMEOUT=$(( (B - 15) * 60 ))" + echo "SELF_TERMINATE_MINUTES=$(( B + 15 ))" + } >> "$GITHUB_ENV" + echo "role_seconds=$(( (B + 15) * 60 ))" >> "$GITHUB_OUTPUT" + + # Harness (gather, bootstrap-common.sh, leg script) and the benchmarked + # code come from one SHA: target_ref is checked out here and the box rebuilds + # it. Tradeoff: an ad-hoc dispatch against an old ref also runs that ref's + # harness. + - name: Checkout target ref + uses: actions/checkout@v4 + with: + ref: ${{ inputs.target_ref }} + + - name: Resolve target SHA + id: resolve + run: echo "sha=$(git rev-parse HEAD)" >> "$GITHUB_OUTPUT" + + # The runner-side half is `go run /gather`. + - uses: ./.github/actions/setup-go + + - name: Configure AWS via OIDC + uses: aws-actions/configure-aws-credentials@v4 + with: + role-to-assume: ${{ secrets.AWS_GHA_ROLE_ARN }} + aws-region: ${{ env.AWS_REGION }} + role-duration-seconds: ${{ steps.deadlines.outputs.role_seconds }} + + - name: Render user-data + # The scripts ship verbatim; parameters travel in a preamble so the bytes + # that run on the box match the bytes in git. Order is: + # preamble exports -> bootstrap-common.sh -> the leg run script. + env: + EXTRA_ENV: ${{ inputs.extra_env }} + run: | + { + 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" + if [ -n "$EXTRA_ENV" ]; then printf '%s\n' "$EXTRA_ENV"; fi + cat "$PERF_EVAL_DIR/bootstrap-common.sh" + cat "$LEG_SCRIPT" + } > /tmp/user-data.sh + + - name: Launch EC2 instance + id: launch + env: + TARGET_REF: ${{ inputs.target_ref }} + RUN_LABEL: ${{ inputs.run_label }} + run: | + COMMON_TAGS="{Key=$TEST_TAG_KEY,Value=$TEST_TAG_VAL}, + {Key=ref,Value=$TARGET_REF}, + {Key=sha,Value=${{ steps.resolve.outputs.sha }}}, + {Key=run-id,Value=${{ github.run_id }}}" + # AMI: Ubuntu 22.04 amd64 in us-east-1 as of 2026-07-08 + RUN_INSTANCES_JSON=$(aws ec2 run-instances \ + --image-id ami-0d28727121d5d4a3c \ + --instance-type "$INSTANCE_TYPE" \ + --iam-instance-profile "Name=$INSTANCE_PROFILE" \ + --user-data file:///tmp/user-data.sh \ + --instance-initiated-shutdown-behavior terminate \ + --block-device-mappings "[{ + \"DeviceName\":\"/dev/sda1\", + \"Ebs\":{\"VolumeSize\":$ROOT_VOLUME_GB,\"VolumeType\":\"gp3\",\"Iops\":$BOOTSTRAP_VOLUME_IOPS,\"Throughput\":$BOOTSTRAP_VOLUME_THROUGHPUT,\"DeleteOnTermination\":true} + }]" \ + --tag-specifications \ + "ResourceType=instance,Tags=[ + {Key=Name,Value=perf-eval-$RUN_LABEL}, + $COMMON_TAGS + ]" \ + "ResourceType=volume,Tags=[ + {Key=Name,Value=perf-eval-${RUN_LABEL}-root}, + $COMMON_TAGS + ]" \ + --count 1 \ + --output json) + + INSTANCE_ID=$(printf '%s' "$RUN_INSTANCES_JSON" | jq -r '.Instances[0].InstanceId') + echo "instance_id=$INSTANCE_ID" >> "$GITHUB_OUTPUT" + + - name: Wait for SSM agent to register + env: + INSTANCE_ID: ${{ steps.launch.outputs.instance_id }} + run: | + DEADLINE=$(( $(date +%s) + SSM_REGISTRATION_TIMEOUT )) + while [ "$(date +%s)" -lt "$DEADLINE" ]; do + PING=$(aws ssm describe-instance-information \ + --filters "Key=InstanceIds,Values=$INSTANCE_ID" \ + --query 'InstanceInformationList[0].PingStatus' \ + --output text 2>/dev/null || echo "") + echo "[$(date -u +%FT%TZ)] ssm ping=$PING" + if [ "$PING" = "Online" ]; then + exit 0 + fi + sleep 10 + done + echo "::error::SSM agent never registered for $INSTANCE_ID — verify AmazonSSMManagedInstanceCore is attached to the stellar-rpc-ci-load-test role" + exit 1 + + # Polls S3 for the result object and relays verdict/passed + the result + # location (bucket/key) as step outputs the coordinator reads. + # RUN_ID must match the box preamble's: gather skips result objects from + # other attempts (a re-run shares RESULT_KEY with its predecessors). + - name: Poll for results + id: results + env: + INSTANCE_ID: ${{ steps.launch.outputs.instance_id }} + RUN_ID: ${{ github.run_id }}-${{ github.run_attempt }} + run: go run "./$PERF_EVAL_DIR/gather" + + - name: Write results summary + if: always() + run: | + if [ -f /tmp/results.md ]; then + cat /tmp/results.md >> "$GITHUB_STEP_SUMMARY" + elif [ -f /tmp/timeout-comment.md ]; then + cat /tmp/timeout-comment.md >> "$GITHUB_STEP_SUMMARY" + fi + + # Standalone-debugging convenience: the result object also lives in S3, which + # is what the coordinator reads. + - name: Upload leg results artifact + if: always() + uses: actions/upload-artifact@v4 + with: + name: perf-eval-results-${{ inputs.run_label }} + path: | + /tmp/results.md + /tmp/timeout-comment.md + /tmp/bench-results.json + if-no-files-found: warn + + - name: Fail leg on timeout or leg failure + if: always() + run: | + if [ "${{ steps.results.outputs.found }}" != "true" ]; then + echo "Leg timed out before producing instance results" + exit 1 + fi + + if [ "${{ steps.results.outputs.passed }}" != "true" ]; then + echo "Instance reported a failing verdict" + cat /tmp/results.md 2>/dev/null || true + exit 1 + fi + + - name: Terminate instance + if: always() && steps.launch.outputs.instance_id != '' + run: | + aws ec2 terminate-instances \ + --instance-ids ${{ steps.launch.outputs.instance_id }} || true diff --git a/.github/workflows/load-test-coordinator.yml b/.github/workflows/load-test-coordinator.yml new file mode 100644 index 000000000..5f66aaa50 --- /dev/null +++ b/.github/workflows/load-test-coordinator.yml @@ -0,0 +1,160 @@ +name: Load test coordinator +# Drives the ephemeral load test(s) for a release: resolves the release context, +# fans out one ec2-leg call per planned leg, then aggregates one consolidated +# sticky comment on the release PR. Owns reporting of perf eval results to the PR. +# Legs report through S3 at runs///result.json; adding a leg +# is one roster entry in the plan job plus its box script. + +defaults: + run: + shell: bash + +on: + push: + branches: [release/**] + workflow_dispatch: + inputs: + target_ref: + description: 'Ref/branch/SHA to load test (default: triggering ref)' + required: false + type: string + default: '' + +# Keyed on the release ref so a new push to the same branch supersedes an +# in-flight test. Each leg's self-terminate ceiling backstops its box. +concurrency: + group: ${{ github.workflow }}-${{ inputs.target_ref || github.ref }} + cancel-in-progress: true + +# Held here and granted to every leg (secrets: inherit) + the report job +permissions: + id-token: write + contents: read + pull-requests: write + +jobs: + plan: + name: Resolve release context + runs-on: ubuntu-latest + timeout-minutes: 5 + outputs: + target_sha: ${{ steps.ctx.outputs.target_sha }} + pr_number: ${{ steps.ctx.outputs.pr_number }} + legs: ${{ steps.ctx.outputs.legs }} + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ inputs.target_ref || github.ref }} + - name: Resolve context + id: ctx + env: + GH_TOKEN: ${{ github.token }} + REF: ${{ inputs.target_ref || github.ref_name }} + run: | + SHA="$(git rev-parse HEAD)" + PR="$(gh pr list --repo "${{ github.repository }}" --state open \ + --base main --head "$REF" --json number --jq '.[0].number // ""' 2>/dev/null || true)" + { + echo "target_sha=$SHA" + echo "pr_number=$PR" + echo 'legs=[{"label":"Apply-load ingestion","run_label":"apply-load","script":"cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/run-load-test.sh"}]' + } >> "$GITHUB_OUTPUT" + + # The fan-out: one matrix entry per 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) }} + uses: ./.github/workflows/ec2-leg.yml + secrets: inherit + with: + target_ref: ${{ needs.plan.outputs.target_sha }} + run_label: ${{ matrix.leg.run_label }} + leg_script: ${{ matrix.leg.script }} + + report: + name: Aggregate + report + needs: [plan, load-test] + if: ${{ !cancelled() }} + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + # Default checkout (github.sha): run renderer at workflow version even + # when the legs benchmark another target_sha. + - uses: actions/checkout@v4 + - uses: ./.github/actions/setup-go + + - name: Configure AWS via OIDC + uses: aws-actions/configure-aws-credentials@v4 + with: + role-to-assume: ${{ secrets.AWS_GHA_ROLE_ARN }} + aws-region: us-east-1 + + # coordinator-runner does the rendering + history fold (it fetches each leg's + # S3 result and emits the new comment body, under unit test). the gh api calls + # stay here. Renderer failures abort before posting. + - name: Render + post sticky report + env: + GH_TOKEN: ${{ github.token }} + PR: ${{ needs.plan.outputs.pr_number }} + REPO: ${{ github.repository }} + AWS_REGION: us-east-1 + TARGET_SHA: ${{ needs.plan.outputs.target_sha }} + TARGET_REF: ${{ inputs.target_ref || github.ref_name }} + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + PLANNED_LEGS: ${{ needs.plan.outputs.legs }} + BUCKET: stellar-rpc-ci-load-test # must match ec2-leg.yml + run: | + set -uo pipefail + MARKER='' + PERF_EVAL=./cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval + + # Read the existing sticky comment (id + body), if any. + CID=""; PREV="" + if [ -n "$PR" ]; then + CID=$(gh api "repos/$REPO/issues/$PR/comments" --paginate \ + --jq "map(select(.user.login == \"github-actions[bot]\" and (.body | startswith(\"$MARKER\")))) | .[0].id // empty" || true) + [ -n "$CID" ] && PREV=$(gh api "repos/$REPO/issues/comments/$CID" --jq '.body' || true) + fi + + # Note result locations are deterministic (runs///result.json) + LEGS=$(jq -c --arg pfx "runs/$GITHUB_RUN_ID" --arg b "$BUCKET" \ + 'map({label, bucket: $b, key: ($pfx + "/" + .run_label + "/result.json")})' <<<"$PLANNED_LEGS") + export LEGS + + # Render the new body (fetches each leg's S3 result, folds prior runs). + if ! printf '%s' "$PREV" | go run "$PERF_EVAL" > /tmp/comment.md; then + echo "::error::renderer failed; leaving sticky comment untouched" + exit 1 + fi + cat /tmp/comment.md >> "$GITHUB_STEP_SUMMARY" + + # Post / edit the sticky comment (skip if the render produced nothing). + if [ ! -s /tmp/comment.md ]; then + echo "::warning::render produced no output, skipping comment post" + elif [ -n "$PR" ]; then + jq -Rs '{body: .}' /tmp/comment.md > /tmp/payload.json + if [ -n "$CID" ]; then + gh api -X PATCH "repos/$REPO/issues/comments/$CID" --input /tmp/payload.json >/dev/null \ + && echo "updated sticky comment $CID on PR #$PR" \ + || echo "::warning::failed to update sticky comment on PR #$PR" + else + gh api -X POST "repos/$REPO/issues/$PR/comments" --input /tmp/payload.json >/dev/null \ + && echo "created sticky comment on PR #$PR" \ + || echo "::warning::failed to create sticky comment on PR #$PR" + fi + fi + + # Authoritative gate: reflects the load-test legs, independent of whether + # the comment posted. + - name: Set overall status + if: always() + run: | + if [ "${{ needs.load-test.result }}" != "success" ]; then + echo "::error::one or more legs did not pass (aggregate result: ${{ needs.load-test.result }})" + exit 1 + fi + echo "all load-test legs passed" diff --git a/.github/workflows/load-test.yml b/.github/workflows/load-test.yml deleted file mode 100644 index 23ea60340..000000000 --- a/.github/workflows/load-test.yml +++ /dev/null @@ -1,216 +0,0 @@ -name: Load test (ephemeral) -# Launches a c5.2xlarge in Horizon (203618453975), polls S3 for the result object -# the box publishes, posts results to the PR, terminates. Box bootstrap lives in -# run-load-test.sh; runner-side polling in runner/orchestrate.go. - -on: - workflow_call: - -permissions: - id-token: write # for OIDC AssumeRole into the GHA role - contents: read - pull-requests: write - -jobs: - load-test: - name: Launch + await ephemeral load-test box - runs-on: ubuntu-latest - timeout-minutes: 225 # 210min results wait + buffer for boot/SSM/poll latency and cleanup (role lasts 240min) - env: - AWS_REGION: us-east-1 - INSTANCE_TYPE: c5.2xlarge - ROOT_VOLUME_GB: 500 - BOOTSTRAP_VOLUME_IOPS: 3000 - # 3000 IOPS is the gp3 floor; 125 MiB/s alone would need only 500. - BOOTSTRAP_VOLUME_THROUGHPUT: 125 - INSTANCE_PROFILE: stellar-rpc-ci-load-test - TEST_TAG_KEY: test - TEST_TAG_VAL: stellar-rpc-ci-load-test - SSM_REGISTRATION_TIMEOUT: 240 # SSM agent registers ~30-90s after boot - RESULTS_TIMEOUT: 12600 # 210 min wait for /tmp/done: ~55m bootstrap+build + ~90m benchmark, under the 170m go-test budget. - POLL_INTERVAL: 30 - DEBUG_LOG_LINES: 40 - DEBUG_LOG_EVERY_POLLS: 5 - LOAD_TEST_DIR: cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test - - steps: - - name: Resolve target context - id: target - env: - GH_TOKEN: ${{ github.token }} - run: | - PR_NUMBER=$(gh pr list \ - --repo "${{ github.repository }}" \ - --state open \ - --base main \ - --head "${{ github.ref_name }}" \ - --json number \ - --jq '.[0].number // ""' 2>/dev/null || true) - - RUN_LABEL="${PR_NUMBER:+pr$PR_NUMBER}" - { - echo "pr_number=$PR_NUMBER" - echo "pr_tag_value=${PR_NUMBER:-none}" - echo "run_label=${RUN_LABEL:-${{ github.ref_name }}}" - } >> "$GITHUB_OUTPUT" - - - name: Checkout target ref - uses: actions/checkout@v4 - with: - ref: ${{ github.sha }} - - # The runner-side half is `go run ... runner orchestrate`. - - uses: ./.github/actions/setup-go - - - name: Configure AWS via OIDC - uses: aws-actions/configure-aws-credentials@v4 - with: - role-to-assume: ${{ secrets.AWS_GHA_ROLE_ARN }} - aws-region: ${{ env.AWS_REGION }} - role-duration-seconds: 14400 - - - name: Resolve latest Ubuntu 22.04 AMI - id: ami - run: | - AMI=$(aws ec2 describe-images \ - --owners 099720109477 \ - --filters \ - "Name=name,Values=ubuntu/images/hvm-ssd*/ubuntu-jammy-22.04-amd64-server-*" \ - "Name=architecture,Values=x86_64" \ - "Name=state,Values=available" \ - --query 'sort_by(Images, &CreationDate)[-1].ImageId' \ - --output text) - echo "ami=$AMI" >> "$GITHUB_OUTPUT" - - - name: Render user-data - # The script ships verbatim; parameters travel in a two-line preamble - # so the bytes that run on the box match the bytes in git. - run: | - { - echo '#!/usr/bin/env bash' - echo 'export TARGET_SHA=${{ github.sha }} RUN_ID=${{ github.run_id }}-${{ github.run_attempt }}' - echo 'export BUCKET=stellar-rpc-ci-load-test RESULT_KEY=runs/${{ github.run_id }}-${{ github.run_attempt }}/result.json' - cat "$LOAD_TEST_DIR/run-load-test.sh" - } > /tmp/user-data.sh - - - name: Launch EC2 instance - id: launch - run: | - COMMON_TAGS="{Key=$TEST_TAG_KEY,Value=$TEST_TAG_VAL}, - {Key=pr,Value=${{ steps.target.outputs.pr_tag_value }}}, - {Key=ref,Value=${{ github.ref_name }}}, - {Key=sha,Value=${{ github.sha }}}, - {Key=run-id,Value=${{ github.run_id }}}" - RUN_INSTANCES_JSON=$(aws ec2 run-instances \ - --image-id "${{ steps.ami.outputs.ami }}" \ - --instance-type "$INSTANCE_TYPE" \ - --iam-instance-profile "Name=$INSTANCE_PROFILE" \ - --user-data file:///tmp/user-data.sh \ - --instance-initiated-shutdown-behavior terminate \ - --block-device-mappings "[{ - \"DeviceName\":\"/dev/sda1\", - \"Ebs\":{\"VolumeSize\":$ROOT_VOLUME_GB,\"VolumeType\":\"gp3\",\"Iops\":$BOOTSTRAP_VOLUME_IOPS,\"Throughput\":$BOOTSTRAP_VOLUME_THROUGHPUT,\"DeleteOnTermination\":true} - }]" \ - --tag-specifications \ - "ResourceType=instance,Tags=[ - {Key=Name,Value=load-test-${{ steps.target.outputs.run_label }}}, - $COMMON_TAGS - ]" \ - "ResourceType=volume,Tags=[ - {Key=Name,Value=load-test-${{ steps.target.outputs.run_label }}-root}, - $COMMON_TAGS - ]" \ - --count 1 \ - --output json) - - INSTANCE_ID=$(printf '%s' "$RUN_INSTANCES_JSON" | jq -r '.Instances[0].InstanceId') - echo "instance_id=$INSTANCE_ID" >> "$GITHUB_OUTPUT" - - - name: Acknowledge launch in PR - if: steps.target.outputs.pr_number != '' - env: - GH_TOKEN: ${{ github.token }} - run: | - if ! gh pr comment ${{ steps.target.outputs.pr_number }} \ - --repo ${{ github.repository }} \ - --body "⏳ Load test launching on \`${{ steps.launch.outputs.instance_id }}\` (commit \`${{ github.sha }}\`). - Workflow run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} - Posting results when the run finishes."; then - echo "::warning::Failed to post launch comment to PR #${{ steps.target.outputs.pr_number }}" - fi - - - name: Wait for SSM agent to register - env: - INSTANCE_ID: ${{ steps.launch.outputs.instance_id }} - run: | - DEADLINE=$(( $(date +%s) + SSM_REGISTRATION_TIMEOUT )) - while [ $(date +%s) -lt $DEADLINE ]; do - PING=$(aws ssm describe-instance-information \ - --filters "Key=InstanceIds,Values=$INSTANCE_ID" \ - --query 'InstanceInformationList[0].PingStatus' \ - --output text 2>/dev/null || echo "") - echo "[$(date -u +%FT%TZ)] ssm ping=$PING" - if [ "$PING" = "Online" ]; then - exit 0 - fi - sleep 10 - done - echo "::error::SSM agent never registered for $INSTANCE_ID — verify AmazonSSMManagedInstanceCore is attached to the stellar-rpc-ci-load-test role" - exit 1 - - - name: Poll for results - id: results - env: - INSTANCE_ID: ${{ steps.launch.outputs.instance_id }} - BUCKET: stellar-rpc-ci-load-test - RESULT_KEY: runs/${{ github.run_id }}-${{ github.run_attempt }}/result.json - run: go run "./$LOAD_TEST_DIR/runner" orchestrate - - - name: Write results summary - if: always() - run: | - if [ -f /tmp/results.md ]; then - cat /tmp/results.md >> "$GITHUB_STEP_SUMMARY" - elif [ -f /tmp/timeout-comment.md ]; then - cat /tmp/timeout-comment.md >> "$GITHUB_STEP_SUMMARY" - fi - - - name: Post results to PR - if: steps.target.outputs.pr_number != '' - env: - GH_TOKEN: ${{ github.token }} - run: | - if [ "${{ steps.results.outputs.found }}" = "true" ]; then - BODY=/tmp/results.md - else - BODY=/tmp/timeout-comment.md - fi - if [ ! -s "$BODY" ]; then - echo "::warning::No body to post to PR #${{ steps.target.outputs.pr_number }} ($BODY missing or empty)" - exit 0 - fi - if ! gh pr comment ${{ steps.target.outputs.pr_number }} \ - --repo ${{ github.repository }} \ - --body-file "$BODY"; then - echo "::warning::Failed to post comment to PR #${{ steps.target.outputs.pr_number }}" - fi - - - name: Fail workflow on timeout or load-test failure - if: always() - run: | - if [ "${{ steps.results.outputs.found }}" != "true" ]; then - echo "Load test timed out before producing instance results" - exit 1 - fi - - if [ "${{ steps.results.outputs.passed }}" != "true" ]; then - echo "Instance reported a failing verdict" - cat /tmp/results.md 2>/dev/null || true - exit 1 - fi - - - name: Terminate instance - if: always() && steps.launch.outputs.instance_id != '' - run: | - aws ec2 terminate-instances \ - --instance-ids ${{ steps.launch.outputs.instance_id }} || true diff --git a/.gitignore b/.gitignore index 3978ca2bd..1c0ebb810 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,4 @@ target/ captive-core/ .soroban/ !test.toml -*.sqlite* +*.sqlite* \ No newline at end of file diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/run-load-test.sh b/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/run-load-test.sh deleted file mode 100755 index 3fb389458..000000000 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/run-load-test.sh +++ /dev/null @@ -1,123 +0,0 @@ -#!/usr/bin/env bash -# Bootstraps the ephemeral load-test box (EC2 user-data): installs the toolchain, -# checks out TARGET_SHA, then hands off to `runner instantiate`, which streams the -# corpus from S3 and runs the ingest benchmark. -# The other half, `runner orchestrate`, polls S3 for the result object. -# -# Result protocol: the box publishes one object to s3://$BUCKET/$RESULT_KEY holding -# {schemaVersion, verdict, markdown, bench, runId, targetSha}. The Go runner -# publishes the success object; this script publishes the fail object so the -# orchestrator always sees a verdict. - -set -euo pipefail -log() { echo "[$(date -u +%FT%TZ)] $*"; } - -exec > >(tee -a /var/log/user-data.log | logger -t user-data -s 2>/dev/console) 2>&1 - -# Hard self-terminate ceiling, independent of the GHA runner for if the runner is -# force-cancelled or crashes and skips its Terminate step. -# Sits above the workflow's 225min job timeout so it never pre-empts a healthy run. -SELF_TERMINATE_MINUTES="${SELF_TERMINATE_MINUTES:-240}" -shutdown -P "+${SELF_TERMINATE_MINUTES}" "load-test self-terminate ceiling" \ - || log "WARN: could not schedule self-terminate ceiling" - -TARGET_SHA="${TARGET_SHA:-}" -RUN_ID="${RUN_ID:-manual}" -REPO="${REPO:-stellar/stellar-rpc}" -WORK_DIR="${WORK_DIR:-/data}" -RESULTS_FILE="${RESULTS_FILE:-/tmp/results.md}" -BUCKET="${BUCKET:-stellar-rpc-ci-load-test}" -RESULT_KEY="${RESULT_KEY:-}" -DEFAULT_BRANCH=main - -# Install the AWS CLI + jq early so any later failure can still publish a result to S3. -log "installing aws cli + jq" -export DEBIAN_FRONTEND=noninteractive -apt-get update -qq -apt-get install -y -qq --no-install-recommends awscli jq curl ca-certificates - -# upload_result publishes {verdict, markdown} as the run's result object so the -# orchestrator (polling S3) sees a verdict. Covers the fail paths. -upload_result() { - local verdict="$1" body_file="$2" - if [ -z "$BUCKET" ] || [ -z "$RESULT_KEY" ]; then - log "WARN: BUCKET/RESULT_KEY unset; cannot publish $verdict result" - return 0 - fi - [ -s "$body_file" ] || printf 'Load test failed before producing a result body.\n' > "$body_file" - jq -n --arg v "$verdict" --rawfile md "$body_file" \ - --arg run "$RUN_ID" --arg sha "$TARGET_SHA" \ - '{schemaVersion: 1, verdict: $v, markdown: $md, runId: $run, targetSha: $sha}' > /tmp/result.json - aws s3api put-object --bucket "$BUCKET" --key "$RESULT_KEY" \ - --content-type application/json --body /tmp/result.json >/dev/null -} - -# bail publishes a fail result the orchestrator can read, then stops. It guards -# only the pre-Go bootstrap phase. -bail() { - log "FATAL: $*" - { printf '❌ **Ingest load test failed** (run %s on `%s`)\n\n```\n' "$RUN_ID" "$TARGET_SHA" - printf '%s\n' "$*" - printf '```\n'; } > "$RESULTS_FILE" - upload_result fail "$RESULTS_FILE" || log "WARN: fail result upload failed" - exit 1 -} -trap 'bail "unhandled error at line $LINENO while running: $BASH_COMMAND"' ERR - -log "clearing stale run state" -rm -f /tmp/bench-results.json /tmp/load-test-ledgers-*.xdr.zstd \ - "$RESULTS_FILE" -rm -rf "$WORK_DIR/stellar-rpc" - -log "installing build deps" -apt-get install -y -qq --no-install-recommends \ - git build-essential \ - libpq5 libsodium23 libunwind8 libc++1-14 - -GO_VERSION=1.25.11 -curl -fsSL "https://go.dev/dl/go${GO_VERSION}.linux-amd64.tar.gz" | tar -xz -C /usr/local -export HOME="${HOME:-/root}" -export GOPATH="${GOPATH:-$HOME/go}" -export GOMODCACHE="${GOMODCACHE:-$GOPATH/pkg/mod}" -export GOCACHE="${GOCACHE:-$HOME/.cache/go-build}" -export CARGO_HOME=/root/.cargo -export RUSTUP_HOME=/root/.rustup -export PATH="/usr/local/go/bin:${CARGO_HOME}/bin:$PATH" -mkdir -p "$GOMODCACHE" "$GOCACHE" "$GOPATH/bin" - -command -v cargo >/dev/null || curl -fsSL https://sh.rustup.rs \ - | sh -s -- -y --profile minimal --default-toolchain stable - -# build-libs needs real git metadata, so this is a git tree/not a source archive. -mkdir -p "$WORK_DIR" -cd "$WORK_DIR" -mkdir -p stellar-rpc && cd stellar-rpc -git init -q -git remote add origin "https://github.com/$REPO.git" -if [ -z "$TARGET_SHA" ]; then - log "TARGET_SHA unset; shallow fetching origin/$DEFAULT_BRANCH" - git fetch --depth 1 origin "$DEFAULT_BRANCH" || bail "failed to fetch origin/$DEFAULT_BRANCH" - git checkout --detach FETCH_HEAD - TARGET_SHA=$(git rev-parse HEAD) -elif git fetch --depth 1 origin "$TARGET_SHA"; then - git checkout --detach FETCH_HEAD -else - log "direct commit fetch failed; falling back to full clone" - cd "$WORK_DIR" && rm -rf stellar-rpc - git clone "https://github.com/$REPO.git" stellar-rpc && cd stellar-rpc - git fetch origin "+refs/pull/*:refs/remotes/origin/pr/*" 2>/dev/null || true - git checkout "$TARGET_SHA" -fi -log "checked out $TARGET_SHA; handing off to the Go runner" - -# The Go runner owns the verdict from here -> release the bootstrap trap. On -# success it publishes the result object itself; on any non-zero exit it has -# written the failure body to RESULTS_FILE (or died before doing so), so we -# publish the fail result it couldn't. -trap - ERR -export TARGET_SHA RUN_ID REPO WORK_DIR RESULTS_FILE BUCKET RESULT_KEY -if ! go run ./cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner instantiate; then - log "go runner exited non-zero; publishing fail result" - upload_result fail "$RESULTS_FILE" - exit 1 -fi diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/instantiate.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/instantiate.go deleted file mode 100644 index 032999d93..000000000 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/instantiate.go +++ /dev/null @@ -1,333 +0,0 @@ -package main - -import ( - "bytes" - "context" - "crypto/sha256" - "encoding/hex" - "encoding/json" - "errors" - "fmt" - "io" - "os" - "os/exec" - "path/filepath" - "strings" - "time" - - "github.com/aws/aws-sdk-go-v2/aws" - "github.com/aws/aws-sdk-go-v2/config" - "github.com/aws/aws-sdk-go-v2/service/s3" - "github.com/klauspost/compress/zstd" -) - -// result is the structured run outcome the box publishes to S3 as an atomic object -// The orchestrator polls for it and reads either a complete object or a 404. -type result struct { - SchemaVersion int `json:"schemaVersion"` - Verdict string `json:"verdict"` // "ok" or "fail" - Markdown string `json:"markdown"` - Bench json.RawMessage `json:"bench,omitempty"` - RunID string `json:"runId"` - TargetSHA string `json:"targetSha"` -} - -func env(key, def string) string { - if v := os.Getenv(key); v != "" { - return v - } - return def -} - -// ledgerScenarios are the apply-load profiles ingested as one concatenated stream, -// one bundle per scenario (load-test-ledgers--.xdr.zstd). -var ( - ledgerScenarios = []string{"oz", "sac", "soroswap"} - curVersion = "v27" // version of the above bundles -) - -// instantiate is the instance half after the bootstrap, which streams the corpus -// from S3, runs the benchmark, and writes the ok/fail verdict. -func instantiate(ctx context.Context) error { - var ( - bucket = env("BUCKET", "stellar-rpc-ci-load-test") - region = env("REGION", "us-east-1") - workDir = env("WORK_DIR", "/data") - goldenDB = env("GOLDEN_DB", filepath.Join(workDir, "golden.sqlite")) - resultsFile = env("RESULTS_FILE", "/tmp/results.md") - benchResults = env("BENCH_RESULTS", "/tmp/bench-results.json") - resultKey = os.Getenv("RESULT_KEY") - targetSHA = os.Getenv("TARGET_SHA") - runID = env("RUN_ID", "manual") - ) - - repoRoot, err := os.Getwd() - if err != nil { - return err - } - bail := func(format string, args ...any) error { - return bailInstance(resultsFile, runID, targetSHA, fmt.Sprintf(format, args...)) - } - - awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region)) - if err != nil { - return bail("loading AWS config: %v", err) - } - fetch := &s3Fetcher{client: s3.NewFromConfig(awsCfg), bucket: bucket} - - bundlePaths, goldenFetchSecs, err := fetchCorpus(ctx, fetch, goldenDB) - if err != nil { - return bail("%v", err) - } - - logger.Infof("download complete") - - logger.Infof("building rpc libs") - if err := runStreaming(ctx, repoRoot, nil, 40, "make", "build-libs"); err != nil { - return bail("make build-libs failed: %v", err) - } - - logger.Infof("running ingest perf benchmark") - benchEnv := []string{ - "LOADTEST_INGEST_LEDGER_PATH=" + strings.Join(bundlePaths, ","), - "LOADTEST_INGEST_DEADLINE=" + env("LOADTEST_INGEST_DEADLINE", "150m"), - "LOADTEST_SQLITE_PATH=" + goldenDB, - "PERF_RESULTS_PATH=" + benchResults, - "PERF_RESULTS_MD_PATH=" + resultsFile, - "PERF_TARGET_SHA=" + targetSHA, - "PERF_RUN_ID=" + runID, - "PERF_REPO=" + env("REPO", "stellar/stellar-rpc"), - fmt.Sprintf("PERF_GOLDEN_FETCH_SECONDS=%d", goldenFetchSecs), - "STELLAR_RPC_INTEGRATION_TESTS_ENABLED=true", - } - if err := runStreaming(ctx, repoRoot, benchEnv, 80, - "go", "test", "-run", "TestIngestSyntheticLedgers", "-timeout", "170m", "-v", - "./cmd/stellar-rpc/internal/integrationtest/"); err != nil { - return bail("benchmark failed:\n%v", err) - } - - if fi, err := os.Stat(resultsFile); err != nil || fi.Size() == 0 { - return bail("benchmark succeeded but did not emit %s", resultsFile) - } - logger.Infof("results ready; publishing verdict") - if err := publishResult( - ctx, fetch.client, bucket, resultKey, "ok", runID, targetSHA, resultsFile, benchResults, - ); err != nil { - return bail("publishing result: %v", err) - } - return nil -} - -// bailInstance writes the failure body and exits non-zero for the bash wrapper -func bailInstance(resultsFile, runID, targetSHA, msg string) error { - logger.Error(msg) - body := fmt.Sprintf("❌ **Ingest load test failed** (run %s on `%s`)\n\n```\n%s\n```\n", runID, targetSHA, msg) - _ = os.WriteFile(resultsFile, []byte(body), 0o644) - os.Exit(1) - return nil // unreachable -} - -// publishResult uploads the run result to s3://bucket/key as one atomic object -// that the orchestrator can poll for. -func publishResult( - ctx context.Context, - client *s3.Client, - bucket, key, verdict, runID, targetSHA, mdPath, benchPath string, -) error { - if key == "" { - logger.Infof("RESULT_KEY unset; skipping S3 result publish (verdict: %s)", verdict) - return nil - } - md, err := os.ReadFile(mdPath) - if err != nil { - return fmt.Errorf("reading result markdown %s: %w", mdPath, err) - } - res := result{ - SchemaVersion: 1, - Verdict: verdict, - Markdown: string(md), - RunID: runID, - TargetSHA: targetSHA, - } - if bench, berr := os.ReadFile(benchPath); berr == nil { - res.Bench = json.RawMessage(bench) - } else { - logger.Warnf("bench results %s unavailable: %v", benchPath, berr) - } - body, err := json.Marshal(res) - if err != nil { - return fmt.Errorf("marshaling result: %w", err) - } - logger.Infof("publishing result to s3://%s/%s (verdict: %s, %d bytes)", bucket, key, verdict, len(body)) - if _, err := client.PutObject(ctx, &s3.PutObjectInput{ - Bucket: &bucket, - Key: &key, - Body: bytes.NewReader(body), - ContentType: aws.String("application/json"), - }); err != nil { - return fmt.Errorf("uploading result: %w", err) - } - return nil -} - -// runStreaming runs name in dir (with extra env appended) and streams combined -// output to our log. On failure, the error carries the last tailN lines. -func runStreaming(ctx context.Context, dir string, env []string, tailN int, name string, args ...string) error { - cmd := exec.CommandContext(ctx, name, args...) - cmd.Dir = dir - cmd.Env = append(os.Environ(), env...) - // The full stream goes to stderr (and on to the box's user-data log), but we - // keep a bounded tail in memory for the error. - tail := &tailWriter{max: 64 << 10} - w := io.MultiWriter(os.Stderr, tail) - cmd.Stdout, cmd.Stderr = w, w - if err := cmd.Run(); err != nil { - return fmt.Errorf("%w\n%s", err, lastLines(tail.String(), tailN)) - } - return nil -} - -// tailWriter is a ring buffer-writer that retains the last max bytes written to it. -type tailWriter struct { - max int - buf []byte -} - -func (w *tailWriter) Write(p []byte) (int, error) { - w.buf = append(w.buf, p...) - if len(w.buf) > w.max { - w.buf = w.buf[len(w.buf)-w.max:] - } - return len(p), nil -} - -func (w *tailWriter) String() string { return string(w.buf) } - -func lastLines(s string, n int) string { - lines := strings.Split(strings.TrimRight(s, "\n"), "\n") - if len(lines) > n { - lines = lines[len(lines)-n:] - } - return strings.Join(lines, "\n") -} - -// fetchCorpus streams the golden DB, stellar-core, and ledger bundles from S3, -// returning the bundle paths and the golden DB fetch duration. -func fetchCorpus(ctx context.Context, fetch *s3Fetcher, goldenDB string) ([]string, int, error) { - // current/prev1/prev2 lets a run fall back to an older golden DB snapshot - // while a fresh one is being published. - goldenFetchSecs := -1 - for _, pfx := range []string{"current", "prev1", "prev2"} { - key := pfx + "/golden.sqlite.zst" - logger.Infof("streaming s3://%s/%s", fetch.bucket, key) - start := time.Now() - if err := fetch.fetchVerified(ctx, key, goldenDB, true, "golden DB"); err != nil { - logger.Infof("%v", err) - _ = os.Remove(goldenDB) - continue - } - goldenFetchSecs = int(time.Since(start).Seconds()) - logger.Infof("golden DB ready in %ds", goldenFetchSecs) - break - } - if goldenFetchSecs < 0 { - return nil, 0, errors.New("no golden.sqlite.zst in current/, prev1/, or prev2/") - } - - const corePath = "/usr/local/bin/stellar-core" // fetch pre-built core cached in S3 - if err := fetch.fetchVerified(ctx, "core/stellar-core.zst", corePath, true, "stellar-core"); err != nil { - return nil, 0, err - } - if err := os.Chmod(corePath, 0o755); err != nil { - return nil, 0, fmt.Errorf("chmod stellar-core: %w", err) - } - - var bundlePaths []string - for _, sc := range ledgerScenarios { - bundlePath := fmt.Sprintf("/tmp/load-test-ledgers-%s-%s.xdr.zstd", curVersion, sc) - key := fmt.Sprintf("ledgers/load-test-ledgers-%s-%s.xdr.zstd", curVersion, sc) - if err := fetch.fetchVerified(ctx, key, bundlePath, false, "ledger bundle ("+sc+")"); err != nil { - return nil, 0, err - } - bundlePaths = append(bundlePaths, bundlePath) - } - return bundlePaths, goldenFetchSecs, nil -} - -// s3Fetcher streams objects from one bucket, sha-verifying when the object -// carries sha256-raw metadata. -type s3Fetcher struct { - client *s3.Client - bucket string -} - -// fetchVerified downloads key to dst (zstd-decoding when zstdMode), checking its -// sha256 against the object's sha256-raw metadata when present. -func (f *s3Fetcher) fetchVerified(ctx context.Context, key, dst string, zstdMode bool, label string) error { - expected := f.expectedSHA(ctx, key, label) - logger.Infof("fetching %s", label) - got, err := f.streamObject(ctx, key, dst, zstdMode) - if err != nil { - return fmt.Errorf("failed to download %s: %w", label, err) - } - if expected != "" && expected != got { - return fmt.Errorf("%s hash mismatch: expected %s, got %s", label, expected, got) - } - if expected == "" { - logger.Infof("%s hash computed (unverified) (%s)", label, got) - } else { - logger.Infof("%s hash OK (%s)", label, got) - } - return nil -} - -// expectedSHA returns the object's sha256-raw metadata, or "" if the object or -// the key is absent (caller then fetches unverified). -func (f *s3Fetcher) expectedSHA(ctx context.Context, key, label string) string { - head, err := f.client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &f.bucket, Key: &key}) - if err != nil { - logger.Warnf("head-object failed for s3://%s/%s; fetching %s without checksum", f.bucket, key, label) - return "" - } - // S3 lowercases user-metadata keys; the SDK strips the x-amz-meta- prefix. - if sha := head.Metadata["sha256-raw"]; sha != "" { - return sha - } - logger.Warnf("no sha256-raw on s3://%s/%s; skipping %s checksum", f.bucket, key, label) - return "" -} - -// streamObject downloads key to dst (zstd-decoding when zstdMode) and returns -// the sha256 of the bytes written. -func (f *s3Fetcher) streamObject(ctx context.Context, key, dst string, zstdMode bool) (string, error) { - out, err := f.client.GetObject(ctx, &s3.GetObjectInput{Bucket: &f.bucket, Key: &key}) - if err != nil { - return "", err - } - defer out.Body.Close() - - var src io.Reader = out.Body - if zstdMode { - zr, err := zstd.NewReader(out.Body) - if err != nil { - return "", err - } - defer zr.Close() - src = zr - } - - file, err := os.Create(dst) - if err != nil { - return "", err - } - defer file.Close() - - hasher := sha256.New() - if _, err := io.Copy(io.MultiWriter(file, hasher), src); err != nil { - return "", err - } - if err := file.Sync(); err != nil { - return "", err - } - return hex.EncodeToString(hasher.Sum(nil)), nil -} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/bootstrap-common.sh b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/bootstrap-common.sh new file mode 100644 index 000000000..3dac90f5b --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/bootstrap-common.sh @@ -0,0 +1,137 @@ +# Generic EC2-leg bootstrap, shared by every perf-eval leg. It is NOT run on its +# own: ec2-leg.yml renders the box user-data as +# + bootstrap-common.sh + run-.sh +# so this file has no shebang. It installs the toolchain and exposes helpers; the +# leg script calls bootstrap_box then hands off to its runner via run_leg. +# +# Result protocol: the box publishes one object to s3://$BUCKET/$RESULT_KEY holding +# {schemaVersion, verdict, markdown, bench, runId, targetSha}. The Go runner +# publishes the success object; bail() here publishes the fail object so the +# gatherer always sees a verdict. + +set -euo pipefail +log() { echo "[$(date -u +%FT%TZ)] $*"; } + +exec > >(tee -a /var/log/user-data.log | logger -t user-data -s 2>/dev/console) 2>&1 + +# Hard self-terminate ceiling, independent of the GHA runner for if the runner is +# force-cancelled or crashes and skips its Terminate step. +# Sits above the workflow's job timeout so it never pre-empts a healthy run. +SELF_TERMINATE_MINUTES="${SELF_TERMINATE_MINUTES:-240}" +shutdown -P "+${SELF_TERMINATE_MINUTES}" "load-test self-terminate ceiling" \ + || log "WARN: could not schedule self-terminate ceiling" + +TARGET_SHA="${TARGET_SHA:-}" +RUN_ID="${RUN_ID:-manual}" +REPO="${REPO:-stellar/stellar-rpc}" +WORK_DIR="${WORK_DIR:-/data}" +RESULTS_FILE="${RESULTS_FILE:-/tmp/results.md}" +BUCKET="${BUCKET:-stellar-rpc-ci-load-test}" +RESULT_KEY="${RESULT_KEY:-}" +DEFAULT_BRANCH=main + +# Install the AWS CLI + jq early so any later failure can still publish a result to S3. +log "installing aws cli + jq" +export DEBIAN_FRONTEND=noninteractive +apt-get update -qq +apt-get install -y -qq --no-install-recommends awscli jq curl ca-certificates + +# upload_result publishes {verdict, markdown} as the run's result object so the +# gatherer (polling S3) sees a verdict. Covers the fail paths. +upload_result() { + local verdict="$1" body_file="$2" + if [ -z "$BUCKET" ] || [ -z "$RESULT_KEY" ]; then + log "WARN: BUCKET/RESULT_KEY unset; cannot publish $verdict result" + return 0 + fi + [ -s "$body_file" ] || printf 'Load test failed before producing a result body.\n' > "$body_file" + jq -n --arg v "$verdict" --rawfile md "$body_file" \ + --arg run "$RUN_ID" --arg sha "$TARGET_SHA" \ + '{schemaVersion: 1, verdict: $v, markdown: $md, runId: $run, targetSha: $sha}' > /tmp/result.json + aws s3api put-object --bucket "$BUCKET" --key "$RESULT_KEY" \ + --content-type application/json --body /tmp/result.json >/dev/null +} + +# bail publishes a fail result the gatherer can read, then stops. It guards +# only the pre-Go bootstrap phase. LEG_TITLE (set by the leg) titles the body. +bail() { + log "FATAL: $*" + { printf '❌ **%s failed** (run %s on `%s`)\n\n```\n' "${LEG_TITLE:-Perf eval}" "$RUN_ID" "$TARGET_SHA" + printf '%s\n' "$*" + printf '```\n'; } > "$RESULTS_FILE" + upload_result fail "$RESULTS_FILE" || log "WARN: fail result upload failed" + exit 1 +} +trap 'bail "unhandled error at line $LINENO while running: $BASH_COMMAND"' ERR + +# persists the full box log to S3 even if instance terminated +upload_box_log() { + [ -n "$BUCKET" ] && [ -n "$RESULT_KEY" ] || return 0 + aws s3 cp /var/log/user-data.log "s3://$BUCKET/${RESULT_KEY%/*}/user-data.log" >/dev/null 2>&1 || true +} +trap upload_box_log EXIT + +# bootstrap_box installs the build toolchain and checks out TARGET_SHA into +# $WORK_DIR/stellar-rpc, leaving the shell cd'd at the repo root. Generic across +# legs; the leg clears any of its own stale artifacts before calling this. +bootstrap_box() { + log "clearing stale run state" + rm -f "$RESULTS_FILE" + rm -rf "$WORK_DIR/stellar-rpc" + + log "installing build deps" + apt-get install -y -qq --no-install-recommends \ + git build-essential \ + libpq5 libsodium23 libunwind8 libc++1-14 + + GO_VERSION=1.25.11 + curl -fsSL "https://go.dev/dl/go${GO_VERSION}.linux-amd64.tar.gz" | tar -xz -C /usr/local + export HOME="${HOME:-/root}" + export GOPATH="${GOPATH:-$HOME/go}" + export GOMODCACHE="${GOMODCACHE:-$GOPATH/pkg/mod}" + export GOCACHE="${GOCACHE:-$HOME/.cache/go-build}" + export CARGO_HOME=/root/.cargo + export RUSTUP_HOME=/root/.rustup + export PATH="/usr/local/go/bin:${CARGO_HOME}/bin:$PATH" + mkdir -p "$GOMODCACHE" "$GOCACHE" "$GOPATH/bin" + + command -v cargo >/dev/null || curl -fsSL https://sh.rustup.rs \ + | sh -s -- -y --profile minimal --default-toolchain stable + + # build-libs needs real git metadata, so this is a git tree/not a source archive. + mkdir -p "$WORK_DIR" + cd "$WORK_DIR" + mkdir -p stellar-rpc && cd stellar-rpc + git init -q + git remote add origin "https://github.com/$REPO.git" + if [ -z "$TARGET_SHA" ]; then + log "TARGET_SHA unset; shallow fetching origin/$DEFAULT_BRANCH" + git fetch --depth 1 origin "$DEFAULT_BRANCH" || bail "failed to fetch origin/$DEFAULT_BRANCH" + git checkout --detach FETCH_HEAD + TARGET_SHA=$(git rev-parse HEAD) + elif git fetch --depth 1 origin "$TARGET_SHA"; then + git checkout --detach FETCH_HEAD + else + log "direct commit fetch failed; falling back to full clone" + cd "$WORK_DIR" && rm -rf stellar-rpc + git clone "https://github.com/$REPO.git" stellar-rpc && cd stellar-rpc + git fetch origin "+refs/pull/*:refs/remotes/origin/pr/*" 2>/dev/null || true + git checkout "$TARGET_SHA" + fi + log "checked out $TARGET_SHA; handing off to the Go runner" +} + +# run_leg hands off to a leg's runner package. The Go runner owns the +# verdict from here, so the bootstrap ERR trap is released first. On a non-zero +# exit the runner has written the failure body to RESULTS_FILE (or died before +# doing so), so we publish the fail result it couldn't. +run_leg() { + local runner_pkg="$1" + trap - ERR + export TARGET_SHA RUN_ID REPO WORK_DIR RESULTS_FILE BUCKET RESULT_KEY + if ! go run "$runner_pkg"; then + log "go runner exited non-zero; publishing fail result" + upload_result fail "$RESULTS_FILE" + exit 1 + fi +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/coordinator-runner.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/coordinator-runner.go new file mode 100644 index 000000000..29c7b5002 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/coordinator-runner.go @@ -0,0 +1,219 @@ +// Command coordinator-runner renders the release perf-eval coordinator's sticky +// PR comment. It reads the prior comment body on stdin, fetches each leg's result +// object from S3, and writes the new comment body to stdout: the current run as +// "Performance Evaluation Test #N" on top, with prior runs folded into drop-downs. +// +// The run history is data that rides inside the comment as a base64 JSON blob in an +// HTML comment. Each render decodes it, prepends the current run, and re-renders +// the whole body from it. +// +// The coordinator workflow (load-test-coordinator.yml) owns the GitHub work: it +// reads the existing comment, pipes it here, and posts/edits what this prints. +package main + +import ( + "cmp" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "os" + "regexp" + "strings" + + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/s3" + + supportlog "github.com/stellar/go-stellar-sdk/support/log" + + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness" +) + +const ( + marker = "" // keys the sticky comment so the workflow can find it + + maxHistory = 10 // runs kept in the history (and so rendered as drop-downs) + maxCommentLen = 60000 // shed oldest runs beyond this; GitHub caps comments at 65536 +) + +var ( + logger = supportlog.New() + historyRe = regexp.MustCompile(`^$`) +) + +// leg is one entry of the LEGS env: where to fetch a leg's result object. +type leg struct { + Label string `json:"label"` + Bucket string `json:"bucket"` + Key string `json:"key"` +} + +// legResult is one leg's outcome as recorded in the history blob. +type legResult struct { + Label string `json:"label"` + Verdict string `json:"verdict"` // "ok"/"fail", empty when no result was published + Markdown string `json:"markdown"` +} + +// runRecord is one coordinator run in the history blob, newest first. +type runRecord struct { + Num int `json:"num"` + TargetSHA string `json:"targetSha"` + TargetRef string `json:"targetRef"` + RunURL string `json:"runUrl"` + Legs []legResult `json:"legs"` +} + +func main() { + logger.SetLevel(supportlog.InfoLevel) + if err := run(context.Background()); err != nil { + logger.Errorf("fatal: %v", err) + os.Exit(1) + } +} + +func run(ctx context.Context) error { + legs, err := parseLegs(os.Getenv("LEGS")) + if err != nil { + return fmt.Errorf("parsing LEGS: %w", err) + } + prev, err := io.ReadAll(os.Stdin) + if err != nil { + return fmt.Errorf("reading previous comment from stdin: %w", err) + } + + awsCfg, err := config.LoadDefaultConfig(ctx, + config.WithRegion(cmp.Or(os.Getenv("AWS_REGION"), "us-east-1"))) + if err != nil { + return err + } + // A missing/unreadable result object (timeout or pre-publish failure) renders + // as a failed leg rather than aborting the report. + s3Client := s3.NewFromConfig(awsCfg) + results := make([]legResult, len(legs)) + for i, l := range legs { + results[i] = legResult{Label: l.Label} + res, err := harness.FetchResult(ctx, s3Client, l.Bucket, l.Key) + if err != nil { + logger.Warnf("no result at s3://%s/%s: %v", l.Bucket, l.Key, err) + continue + } + results[i].Verdict, results[i].Markdown = res.Verdict, res.Markdown + } + + body := renderComment(runRecord{ + TargetSHA: os.Getenv("TARGET_SHA"), + TargetRef: os.Getenv("TARGET_REF"), + RunURL: os.Getenv("RUN_URL"), + Legs: results, + }, string(prev)) + _, err = io.WriteString(os.Stdout, body) + return err +} + +// renderComment numbers cur, prepends it to the prior comment's history, and +// renders the whole body from it, shedding the oldest runs to stay within the +// history and comment-size caps. +func renderComment(cur runRecord, prev string) string { + hist := parseHistory(prev) + cur.Num = 1 + if len(hist) > 0 { + cur.Num = hist[0].Num + 1 + } + hist = append([]runRecord{cur}, hist...) + if len(hist) > maxHistory { + hist = hist[:maxHistory] + } + body := renderBody(hist) + for len(body) > maxCommentLen && len(hist) > 1 { + hist = hist[:len(hist)-1] + body = renderBody(hist) + } + return body +} + +func renderBody(hist []runRecord) string { + var b strings.Builder + fmt.Fprintln(&b, marker) + fmt.Fprintf(&b, "\n", encodeHistory(hist)) + b.WriteString(renderRun(hist[0])) + for _, r := range hist[1:] { + fmt.Fprintf(&b, "\n
\nPerformance Evaluation Test #%d\n\n", r.Num) + b.WriteString(strings.Trim(renderRun(r), "\n")) + b.WriteString("\n
\n") + } + return b.String() +} + +func renderRun(r runRecord) string { + var b strings.Builder + fmt.Fprintf(&b, "## 🧪 Performance Evaluation Test #%d\n\n", r.Num) + fmt.Fprintf(&b, "**Commit:** `%s` (`%s`)\n", r.TargetSHA[:min(12, len(r.TargetSHA))], r.TargetRef) + fmt.Fprintf(&b, "**Run:** %s\n", r.RunURL) + for _, l := range r.Legs { + b.WriteString(renderLeg(l)) + } + return b.String() +} + +// renderLeg renders one leg's section: a verdict heading plus its result markdown, +// or a fallback when no result object was published. +func renderLeg(l legResult) string { + emoji := "❌" + if l.Verdict == "ok" { + emoji = "✅" + } + var b strings.Builder + fmt.Fprintf(&b, "\n### %s %s — verdict: %s\n\n", emoji, l.Label, cmp.Or(l.Verdict, "none")) + if strings.TrimSpace(l.Markdown) != "" { + b.WriteString(strings.TrimRight(l.Markdown, "\n")) + b.WriteByte('\n') + } else { + b.WriteString("_No result object published (leg timed out or failed before publishing). See the run logs._\n") + } + return b.String() +} + +func encodeHistory(hist []runRecord) string { + data, err := json.Marshal(hist) + if err != nil { + logger.Warnf("marshaling history; next run will start fresh: %v", err) + return "" + } + return base64.StdEncoding.EncodeToString(data) +} + +// parseHistory recovers the run history from the blob line of a prior comment +// body. An absent or undecodable blob starts the history fresh. +func parseHistory(prev string) []runRecord { + for line := range strings.SplitSeq(prev, "\n") { + m := historyRe.FindStringSubmatch(line) + if m == nil { + continue + } + data, err := base64.StdEncoding.DecodeString(m[1]) + if err != nil { + logger.Warnf("undecodable history blob; starting fresh: %v", err) + return nil + } + var hist []runRecord + if err := json.Unmarshal(data, &hist); err != nil { + logger.Warnf("unparseable history blob; starting fresh: %v", err) + return nil + } + return hist + } + return nil +} + +func parseLegs(s string) ([]leg, error) { + if strings.TrimSpace(s) == "" { + return nil, nil + } + var legs []leg + if err := json.Unmarshal([]byte(s), &legs); err != nil { + return nil, err + } + return legs, nil +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/coordinator-runner_test.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/coordinator-runner_test.go new file mode 100644 index 000000000..cd28a29d9 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/coordinator-runner_test.go @@ -0,0 +1,121 @@ +package main + +import ( + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func okLeg(md string) []legResult { + return []legResult{{Label: "Apply-load ingestion", Verdict: "ok", Markdown: md}} +} + +func renderN(n int, legs []legResult) string { + var prev string + for i := range n { + prev = renderComment(runRecord{ + TargetSHA: strings.Repeat("a", 12), + TargetRef: "release/v1", + RunURL: fmt.Sprintf("https://example/run/%d", i+1), + Legs: legs, + }, prev) + } + return prev +} + +func TestRenderComment_FirstRun(t *testing.T) { + out := renderComment(runRecord{ + TargetSHA: "abcdef1234567890", + TargetRef: "release/v1.2.3", + RunURL: "https://example/run/1", + Legs: okLeg("throughput: 42 l/s"), + }, "") + require.True(t, strings.HasPrefix(out, marker+"\n")) + require.Contains(t, out, "## 🧪 Performance Evaluation Test #1") + require.Contains(t, out, "**Commit:** `abcdef123456` (`release/v1.2.3`)") + require.Contains(t, out, "### ✅ Apply-load ingestion — verdict: ok") + require.Contains(t, out, "throughput: 42 l/s") + require.NotContains(t, out, "
") // no history on the first run +} + +func TestRenderLeg_FallbackWhenNoResult(t *testing.T) { + out := renderLeg(legResult{Label: "X"}) + require.Contains(t, out, "### ❌ X — verdict: none") + require.Contains(t, out, "No result object published") +} + +// Four runs in sequence: each run's output is the next run's "prior comment". +// The result must keep the current run on top and the rest as drop-downs. +func TestRenderComment_FoldsHistory(t *testing.T) { + out := renderN(4, okLeg("run body")) + + require.Equal(t, 1, strings.Count(out, marker)) + require.Contains(t, out, "## 🧪 Performance Evaluation Test #4") + require.Equal(t, 3, strings.Count(out, "
\nPerformance Evaluation Test #")) + + // Each prior run in descending order (#3, then #2, then #1) + i3 := strings.Index(out, "Performance Evaluation Test #3") + i2 := strings.Index(out, "Performance Evaluation Test #2") + i1 := strings.Index(out, "Performance Evaluation Test #1") + require.Positive(t, i3) + require.Less(t, i3, i2) + require.Less(t, i2, i1) +} + +// Leg markdown containing its own
block must survive folding intact: +// the history is data, so rendered markup is never parsed back. +func TestRenderComment_LegMarkdownWithDetailsSurvivesFold(t *testing.T) { + legMD := "table\n\n
\nper-profile breakdown\n\ninner\n
" + out := renderN(3, okLeg(legMD)) + + hist := parseHistory(out) + require.Len(t, hist, 3) + for _, r := range hist { + require.Equal(t, legMD, r.Legs[0].Markdown) + } + // The inner drop-down renders in every fold without hijacking the structure. + require.Equal(t, 3, strings.Count(out, "per-profile breakdown")) + require.Equal(t, 2, strings.Count(out, "Performance Evaluation Test #")) +} + +// Numbering continues past the history cap; the oldest runs are shed. +func TestRenderComment_CapsHistory(t *testing.T) { + out := renderN(maxHistory+2, okLeg("run body")) + + hist := parseHistory(out) + require.Len(t, hist, maxHistory) + require.Equal(t, maxHistory+2, hist[0].Num) + require.Contains(t, out, fmt.Sprintf("## 🧪 Performance Evaluation Test #%d", maxHistory+2)) + require.NotContains(t, out, "Performance Evaluation Test #2\n") // shed + require.NotContains(t, out, "Performance Evaluation Test #1\n") // shed +} + +// Oversized histories shed old runs to stay under the comment-size cap. +func TestRenderComment_ShedsRunsOverSizeCap(t *testing.T) { + big := strings.Repeat("x", maxCommentLen/3) + out := renderN(5, okLeg(big)) + + require.LessOrEqual(t, len(out), maxCommentLen) + hist := parseHistory(out) + require.NotEmpty(t, hist) + require.Equal(t, 5, hist[0].Num) // current run always kept +} + +// A prior comment without a parseable blob (legacy, corrupt, or hand-edited) +// starts the series fresh instead of failing. +func TestParseHistory_FreshOnAbsentOrCorrupt(t *testing.T) { + require.Nil(t, parseHistory("")) + require.Nil(t, parseHistory(marker+"\n## some legacy comment\n")) + require.Nil(t, parseHistory("")) + require.Nil(t, parseHistory("")) // "not-json" + + out := renderComment(runRecord{ + TargetSHA: strings.Repeat("a", 12), + TargetRef: "release/v1", + RunURL: "https://example/run", + Legs: okLeg("body"), + }, marker+"\n## some legacy comment\n") + require.Contains(t, out, "## 🧪 Performance Evaluation Test #1") +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/gather/main.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/gather/main.go new file mode 100644 index 000000000..d433b4c30 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/gather/main.go @@ -0,0 +1,7 @@ +// Command gather is the GHA-runner half of every perf-eval leg: it polls S3 +// for the result object the box publishes and relays it as step outputs. +package main + +import "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness" + +func main() { harness.Run(harness.Gather) } diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/exec.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/exec.go new file mode 100644 index 000000000..45ddd9551 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/exec.go @@ -0,0 +1,60 @@ +package harness + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "strings" +) + +// RunStreaming runs name in dir (with extra env appended) and streams combined +// output to our log. On failure, the error carries the last tailN lines. +func RunStreaming(ctx context.Context, dir string, env []string, tailN int, name string, args ...string) error { + cmd := exec.CommandContext(ctx, name, args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), env...) + // The full stream goes to stderr (and on to the box's user-data log), but we + // keep a bounded tail in memory for the error. + tail := &tailWriter{max: 64 << 10} + w := io.MultiWriter(os.Stderr, tail) + cmd.Stdout, cmd.Stderr = w, w + if err := cmd.Run(); err != nil { + return fmt.Errorf("%w\n%s", err, lastLines(tail.String(), tailN)) + } + return nil +} + +// BailInstance writes the leg's failure body to resultsFile and returns msg as +// the error for the runner to exit with. +func BailInstance(resultsFile, title, runID, targetSHA, msg string) error { + body := fmt.Sprintf("❌ **%s failed** (run %s on `%s`)\n\n```\n%s\n```\n", title, runID, targetSHA, msg) + _ = os.WriteFile(resultsFile, []byte(body), 0o644) + return errors.New(msg) +} + +// tailWriter is a ring buffer-writer that retains the last max bytes written to it. +type tailWriter struct { + max int + buf []byte +} + +func (w *tailWriter) Write(p []byte) (int, error) { + w.buf = append(w.buf, p...) + if len(w.buf) > w.max { + w.buf = w.buf[len(w.buf)-w.max:] + } + return len(p), nil +} + +func (w *tailWriter) String() string { return string(w.buf) } + +func lastLines(s string, n int) string { + lines := strings.Split(strings.TrimRight(s, "\n"), "\n") + if len(lines) > n { + lines = lines[len(lines)-n:] + } + return strings.Join(lines, "\n") +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/orchestrate.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/gather.go similarity index 55% rename from cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/orchestrate.go rename to cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/gather.go index 4407281f2..759e44b28 100644 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/orchestrate.go +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/gather.go @@ -1,25 +1,9 @@ -// Command runner drives the ephemeral RPC ingestion load test. It has two -// subcommands, one per environment the test spans: -// -// runner instantiate on the EC2 box, after a shell preamble has installed -// the toolchain and checked out the repo: streams the -// golden DB, stellar-core, and ledger bundles from S3 -// (sha-verified), runs the ingest benchmark, and writes -// an ok/fail verdict. -// runner orchestrate on the GHA runner: polls S3 for the result object the -// instance publishes and relays the verdict + results as -// step outputs. SSM carries only best-effort debug tails. -// -// The two halves coordinate through a single S3 result object (see type result). -// SSM is used only for live-progress and timeout diagnostics. -package main +package harness import ( "context" - "encoding/json" "errors" "fmt" - "io" "os" "strconv" "strings" @@ -28,69 +12,24 @@ import ( "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/s3" - "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/aws/aws-sdk-go-v2/service/ssm" - "github.com/aws/smithy-go" - - supportlog "github.com/stellar/go-stellar-sdk/support/log" ) -var logger = supportlog.New() - -func main() { - logger.SetLevel(supportlog.InfoLevel) - - cmd := "instantiate" - if len(os.Args) > 1 { - cmd = os.Args[1] - } - - ctx := context.Background() - var err error - switch cmd { - case "instantiate": - err = instantiate(ctx) - case "orchestrate": - err = orchestrate(ctx) - default: - fmt.Fprintf(os.Stderr, "usage: %s [instantiate|orchestrate]\n", os.Args[0]) - os.Exit(64) - } - if err != nil { - logger.Errorf("fatal: %v", err) - os.Exit(1) - } -} - // commandWaitTimeout backstops a stuck SSM command (the debug-tail reads). const commandWaitTimeout = 60 * time.Second -// 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 -} - -// orchestrate polls the box until it reports a verdict and relays the result as step outputs -// On timeout it writes a debug comment instead. -func orchestrate(ctx context.Context) error { - vals, err := requireEnv("INSTANCE_ID", "AWS_REGION", +// Gather is the GHA-runner half: it polls S3 until the box reports a verdict +// and relays the result as step outputs. On timeout it writes a debug comment +// instead. Used by every leg's runner. +func Gather(ctx context.Context) error { + vals, err := RequireEnv("INSTANCE_ID", "AWS_REGION", "RESULTS_TIMEOUT", "POLL_INTERVAL", "GITHUB_OUTPUT", "DEBUG_LOG_LINES", "DEBUG_LOG_EVERY_POLLS", - "BUCKET", "RESULT_KEY") + "BUCKET", "RESULT_KEY", "RUN_ID") if err != nil { return err } instanceID, region, githubOutput := vals[0], vals[1], vals[4] - bucket, resultKey := vals[7], vals[8] + bucket, resultKey, runID := vals[7], vals[8], vals[9] resultsTimeoutSec, err := strconv.Atoi(vals[2]) if err != nil { @@ -120,12 +59,16 @@ func orchestrate(ctx context.Context) error { deadline := time.Now().Add(resultsTimeout) for pollCount := 1; time.Now().Before(deadline); pollCount++ { - res, derr := fetchResult(ctx, s3Client, bucket, resultKey) + res, derr := FetchResult(ctx, s3Client, bucket, resultKey) switch { - case errors.Is(derr, errResultNotReady): + case errors.Is(derr, ErrResultNotReady): logger.Infof("still waiting for s3://%s/%s", bucket, resultKey) case derr != nil: logger.Warnf("result fetch failed; retrying: %v", derr) + // A leftover object from a prior attempt (re-runs share RESULT_KEY) is + // "not published yet" so this attempt's box overwrites it. + case res.RunID != runID: + logger.Infof("ignoring stale result from run %s (want %s)", res.RunID, runID) default: logger.Infof("result published by instance (verdict: %s)", res.Verdict) if werr := os.WriteFile("/tmp/results.md", []byte(res.Markdown), 0o644); werr != nil { @@ -151,8 +94,7 @@ type ssmRunner struct { instanceID string } -// capture dispatches command, waits for it, and returns its stdout. A non-nil -// error means dispatch failed; an unreadable result is "". +// capture dispatches command, waits for it, and returns its stdout. func (r *ssmRunner) capture(ctx context.Context, command string) (string, error) { var id string var sendErr error @@ -195,59 +137,6 @@ func (r *ssmRunner) debugTail(ctx context.Context, n int) string { return out } -// errResultNotReady means the result object hasn't been published yet. -var errResultNotReady = errors.New("result not published yet") - -// fetchResult gets and decodes the result object, returning errResultNotReady -// when it is absent. -func fetchResult(ctx context.Context, client *s3.Client, bucket, key string) (*result, error) { - out, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: &bucket, Key: &key}) - if err != nil { - if isNotFound(err) { - return nil, errResultNotReady - } - return nil, err - } - defer out.Body.Close() - - data, err := io.ReadAll(out.Body) - if err != nil { - return nil, err - } - var res result - if err := json.Unmarshal(data, &res); err != nil { - return nil, fmt.Errorf("decoding result object: %w", err) - } - return &res, nil -} - -// isNotFound reports whether a GetObject error means the key is absent. -func isNotFound(err error) bool { - var nsk *types.NoSuchKey - if errors.As(err, &nsk) { - return true - } - var apiErr smithy.APIError - if errors.As(err, &apiErr) { - switch apiErr.ErrorCode() { - case "NoSuchKey", "NotFound": - return true - } - } - return false -} - -// 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) - if err != nil { - return err - } - defer f.Close() - _, err = fmt.Fprintln(f, strings.Join(lines, "\n")) - return err -} - // writeTimeoutComment is the no-verdict path: it writes a comment to // /tmp/timeout-comment.md and records found=false. func writeTimeoutComment( 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 new file mode 100644 index 000000000..13d3c7883 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness.go @@ -0,0 +1,77 @@ +// Package harness holds the generic EC2-leg machinery shared by the perf-eval +// legs. Each leg spins up an ephemeral box, bootstraps it identically, runs a +// leg-specific task, and reports back through one S3 result object. +// +// This package owns the parts that are identical across legs: +// +// Gather GHA-side: polls S3 for the result object the box publishes +// and relays the verdict + results as step outputs. +// S3Fetcher on-box: streams (and sha-verifies) corpus objects from S3. +// PublishResult on-box: writes the ok/fail result object the gatherer reads. +// RunStreaming on-box: runs a child, streaming output with a bounded tail. +// +// Leg-specific work (which corpus to fetch, which task to run) lives in each +// leg's own on-box runner command; Gather is its own command (perf-eval/gather) +// shared by all legs. +package harness + +import ( + "context" + "fmt" + "os" + "strings" + + supportlog "github.com/stellar/go-stellar-sdk/support/log" +) + +// NewLogger returns an Info-level logger (supportlog.New starts at WARN). Each +// leg's runner uses one for its own messages. +func NewLogger() *supportlog.Entry { + l := supportlog.New() + l.SetLevel(supportlog.InfoLevel) + return l +} + +var logger = NewLogger() + +// Run executes a command's task, logging and exiting non-zero on error. +func Run(task func(context.Context) error) { + if err := task(context.Background()); err != nil { + logger.Errorf("fatal: %v", err) + os.Exit(1) + } +} + +// Env returns the value of key, or def when unset/empty. +func Env(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + 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 +} + +// 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) + if err != nil { + return err + } + defer f.Close() + _, err = fmt.Fprintln(f, strings.Join(lines, "\n")) + return err +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/runner_test.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness_test.go similarity index 90% rename from cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/runner_test.go rename to cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness_test.go index 33c394d99..7e3c41d68 100644 --- a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/runner/runner_test.go +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/harness_test.go @@ -1,4 +1,4 @@ -package main +package harness import ( "encoding/json" @@ -17,16 +17,16 @@ func TestIsNotFound(t *testing.T) { require.False(t, isNotFound(errors.New("i/o timeout"))) } -// TestResultRoundTrip guards the publisher/poller contract: what publishResult -// writes must decode back to what orchestrate relays. +// TestResultRoundTrip guards the publisher/poller contract: what PublishResult +// writes must decode back to what Gather relays. func TestResultRoundTrip(t *testing.T) { - in := result{ + in := Result{ SchemaVersion: 1, Verdict: "ok", Markdown: "# r", Bench: json.RawMessage(`{"x":1}`), RunID: "123-1", TargetSHA: "abc", } data, err := json.Marshal(in) require.NoError(t, err) - var out result + var out Result require.NoError(t, json.Unmarshal(data, &out)) require.Equal(t, in, out) } diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/s3.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/s3.go new file mode 100644 index 000000000..27ad0ca47 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness/s3.go @@ -0,0 +1,191 @@ +package harness + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/aws/smithy-go" + "github.com/klauspost/compress/zstd" +) + +// Result is the structured run outcome the box publishes to S3 as an atomic +// object. The gatherer polls for it and reads either a complete object or +// a 404. +type Result struct { + SchemaVersion int `json:"schemaVersion"` + Verdict string `json:"verdict"` // "ok" or "fail" + Markdown string `json:"markdown"` + Bench json.RawMessage `json:"bench,omitempty"` + RunID string `json:"runId"` + TargetSHA string `json:"targetSha"` +} + +// PublishResult uploads the run result to s3://bucket/key as one atomic object +// that the gatherer can poll for. +func PublishResult( + ctx context.Context, + client *s3.Client, + bucket, key, verdict, runID, targetSHA, mdPath, benchPath string, +) error { + if key == "" { + logger.Infof("RESULT_KEY unset; skipping S3 result publish (verdict: %s)", verdict) + return nil + } + md, err := os.ReadFile(mdPath) + if err != nil { + return fmt.Errorf("reading result markdown %s: %w", mdPath, err) + } + res := Result{ + SchemaVersion: 1, + Verdict: verdict, + Markdown: string(md), + RunID: runID, + TargetSHA: targetSHA, + } + if benchPath != "" { + if bench, berr := os.ReadFile(benchPath); berr == nil { + res.Bench = json.RawMessage(bench) + } else { + logger.Warnf("bench results %s unavailable: %v", benchPath, berr) // best-effort for local runs + } + } + body, err := json.Marshal(res) + if err != nil { + return fmt.Errorf("marshaling result: %w", err) + } + logger.Infof("publishing result to s3://%s/%s (verdict: %s, %d bytes)", bucket, key, verdict, len(body)) + if _, err := client.PutObject(ctx, &s3.PutObjectInput{ + Bucket: &bucket, + Key: &key, + Body: bytes.NewReader(body), + ContentType: aws.String("application/json"), + }); err != nil { + return fmt.Errorf("uploading result: %w", err) + } + return nil +} + +// ErrResultNotReady means the result object hasn't been published yet. +var ErrResultNotReady = errors.New("result not published yet") + +// FetchResult gets and decodes the result object, returning ErrResultNotReady +// when it is absent. +func FetchResult(ctx context.Context, client *s3.Client, bucket, key string) (*Result, error) { + out, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: &bucket, Key: &key}) + if err != nil { + if isNotFound(err) { + return nil, ErrResultNotReady + } + return nil, err + } + defer out.Body.Close() + + data, err := io.ReadAll(out.Body) + if err != nil { + return nil, err + } + var res Result + if err := json.Unmarshal(data, &res); err != nil { + return nil, fmt.Errorf("decoding result object: %w", err) + } + return &res, nil +} + +// isNotFound reports whether a GetObject error means the key is absent. +func isNotFound(err error) bool { + var nsk *types.NoSuchKey + var apiErr smithy.APIError + if errors.As(err, &nsk) || + errors.As(err, &apiErr) && (apiErr.ErrorCode() == "NoSuchKey" || apiErr.ErrorCode() == "NotFound") { + return true + } + return false +} + +// S3Fetcher streams objects from one bucket, sha-verifying when the object +// carries sha256-raw metadata. +type S3Fetcher struct { + Client *s3.Client + Bucket string +} + +// FetchVerified downloads key to dst (zstd-decoding when zstdMode), checking its +// sha256 against the object's sha256-raw metadata when present. +func (f *S3Fetcher) FetchVerified(ctx context.Context, key, dst string, zstdMode bool, label string) error { + expected := f.expectedSHA(ctx, key, label) + logger.Infof("fetching %s", label) + got, err := f.streamObject(ctx, key, dst, zstdMode) + if err != nil { + return fmt.Errorf("failed to download %s: %w", label, err) + } + if expected != "" && expected != got { + return fmt.Errorf("%s hash mismatch: expected %s, got %s", label, expected, got) + } + if expected == "" { + logger.Infof("%s hash computed (unverified) (%s)", label, got) + } else { + logger.Infof("%s hash OK (%s)", label, got) + } + return nil +} + +// expectedSHA returns the object's sha256-raw metadata, or "" if the object or +// the key is absent (caller then fetches unverified). +func (f *S3Fetcher) expectedSHA(ctx context.Context, key, label string) string { + head, err := f.Client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &f.Bucket, Key: &key}) + if err != nil { + logger.Warnf("head-object failed for s3://%s/%s; fetching %s without checksum", f.Bucket, key, label) + return "" + } + // S3 lowercases user-metadata keys; the SDK strips the x-amz-meta- prefix. + if sha := head.Metadata["sha256-raw"]; sha != "" { + return sha + } + logger.Warnf("no sha256-raw on s3://%s/%s; skipping %s checksum", f.Bucket, key, label) + return "" +} + +// streamObject downloads key to dst (zstd-decoding when zstdMode) and returns +// the sha256 of the bytes written. +func (f *S3Fetcher) streamObject(ctx context.Context, key, dst string, zstdMode bool) (string, error) { + out, err := f.Client.GetObject(ctx, &s3.GetObjectInput{Bucket: &f.Bucket, Key: &key}) + if err != nil { + return "", err + } + defer out.Body.Close() + + var src io.Reader = out.Body + if zstdMode { + zr, err := zstd.NewReader(out.Body) + if err != nil { + return "", err + } + defer zr.Close() + src = zr + } + + file, err := os.Create(dst) + if err != nil { + return "", err + } + defer file.Close() + + hasher := sha256.New() + if _, err := io.Copy(io.MultiWriter(file, hasher), src); err != nil { + return "", err + } + if err := file.Sync(); err != nil { + return "", err + } + return hex.EncodeToString(hasher.Sum(nil)), nil +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/run-load-test.sh b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/run-load-test.sh new file mode 100644 index 000000000..d08fccdb0 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/run-load-test.sh @@ -0,0 +1,11 @@ +# Apply-load 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 the leg runner, which streams the corpus from S3 +# and runs the ingest benchmark. +LEG_TITLE="Ingest load test" + +log "clearing stale apply-load state" +rm -f /tmp/bench-results.json /tmp/load-test-ledgers-*.xdr.zstd + +bootstrap_box +run_leg ./cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/runner diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/runner/instantiate.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/runner/instantiate.go new file mode 100644 index 000000000..c1a21b7d6 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/runner/instantiate.go @@ -0,0 +1,138 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/s3" + + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness" +) + +// ledgerScenarios are the apply-load profiles ingested as one concatenated stream, +// one bundle per scenario (load-test-ledgers--.xdr.zstd). +var ( + ledgerScenarios = []string{"oz", "sac", "soroswap"} + curVersion = "v27" // version of the above bundles +) + +// instantiate is the instance half after the bootstrap, which streams the corpus +// from S3, runs the benchmark, and writes the ok/fail verdict. +func instantiate(ctx context.Context) error { + var ( + bucket = harness.Env("BUCKET", "stellar-rpc-ci-load-test") + region = harness.Env("REGION", "us-east-1") + workDir = harness.Env("WORK_DIR", "/data") + goldenDB = harness.Env("GOLDEN_DB", filepath.Join(workDir, "golden.sqlite")) + resultsFile = harness.Env("RESULTS_FILE", "/tmp/results.md") + benchResults = harness.Env("BENCH_RESULTS", "/tmp/bench-results.json") + resultKey = os.Getenv("RESULT_KEY") + targetSHA = os.Getenv("TARGET_SHA") + runID = harness.Env("RUN_ID", "manual") + ) + + repoRoot, err := os.Getwd() + if err != nil { + return err + } + bail := func(format string, args ...any) error { + return harness.BailInstance(resultsFile, "Ingest load test", runID, targetSHA, fmt.Sprintf(format, args...)) + } + + awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region)) + if err != nil { + return bail("loading AWS config: %v", err) + } + fetch := &harness.S3Fetcher{Client: s3.NewFromConfig(awsCfg), Bucket: bucket} + + bundlePaths, goldenFetchSecs, err := fetchCorpus(ctx, fetch, goldenDB) + if err != nil { + return bail("%v", err) + } + + logger.Infof("download complete") + + logger.Infof("building rpc libs") + if err := harness.RunStreaming(ctx, repoRoot, nil, 40, "make", "build-libs"); err != nil { + return bail("make build-libs failed: %v", err) + } + + logger.Infof("running ingest perf benchmark") + benchEnv := []string{ + "LOADTEST_INGEST_LEDGER_PATH=" + strings.Join(bundlePaths, ","), + "LOADTEST_INGEST_DEADLINE=" + harness.Env("LOADTEST_INGEST_DEADLINE", "150m"), + "LOADTEST_SQLITE_PATH=" + goldenDB, + "PERF_RESULTS_PATH=" + benchResults, + "PERF_RESULTS_MD_PATH=" + resultsFile, + "PERF_TARGET_SHA=" + targetSHA, + "PERF_RUN_ID=" + runID, + "PERF_REPO=" + harness.Env("REPO", "stellar/stellar-rpc"), + fmt.Sprintf("PERF_GOLDEN_FETCH_SECONDS=%d", goldenFetchSecs), + "STELLAR_RPC_INTEGRATION_TESTS_ENABLED=true", + } + if err := harness.RunStreaming(ctx, repoRoot, benchEnv, 80, + "go", "test", "-run", "TestIngestSyntheticLedgers", "-timeout", "170m", "-v", + "./cmd/stellar-rpc/internal/integrationtest/"); err != nil { + return bail("benchmark failed:\n%v", err) + } + + if fi, err := os.Stat(resultsFile); err != nil || fi.Size() == 0 { + return bail("benchmark succeeded but did not emit %s", resultsFile) + } + logger.Infof("results ready; publishing verdict") + if err := harness.PublishResult( + ctx, fetch.Client, bucket, resultKey, "ok", runID, targetSHA, resultsFile, benchResults, + ); err != nil { + return bail("publishing result: %v", err) + } + return nil +} + +// fetchCorpus streams the golden DB, stellar-core, and ledger bundles from S3, +// returning the bundle paths and the golden DB fetch duration. +func fetchCorpus(ctx context.Context, fetch *harness.S3Fetcher, goldenDB string) ([]string, int, error) { + // current/prev1/prev2 lets a run fall back to an older golden DB snapshot + // while a fresh one is being published. + goldenFetchSecs := -1 + for _, pfx := range []string{"current", "prev1", "prev2"} { + key := pfx + "/golden.sqlite.zst" + logger.Infof("streaming s3://%s/%s", fetch.Bucket, key) + start := time.Now() + if err := fetch.FetchVerified(ctx, key, goldenDB, true, "golden DB"); err != nil { + logger.Infof("%v", err) + _ = os.Remove(goldenDB) + continue + } + goldenFetchSecs = int(time.Since(start).Seconds()) + logger.Infof("golden DB ready in %ds", goldenFetchSecs) + break + } + if goldenFetchSecs < 0 { + return nil, 0, errors.New("no golden.sqlite.zst in current/, prev1/, or prev2/") + } + + const corePath = "/usr/local/bin/stellar-core" // fetch pre-built core cached in S3 + if err := fetch.FetchVerified(ctx, "core/stellar-core.zst", corePath, true, "stellar-core"); err != nil { + return nil, 0, err + } + if err := os.Chmod(corePath, 0o755); err != nil { + return nil, 0, fmt.Errorf("chmod stellar-core: %w", err) + } + + var bundlePaths []string + for _, sc := range ledgerScenarios { + bundlePath := fmt.Sprintf("/tmp/load-test-ledgers-%s-%s.xdr.zstd", curVersion, sc) + key := fmt.Sprintf("ledgers/load-test-ledgers-%s-%s.xdr.zstd", curVersion, sc) + if err := fetch.FetchVerified(ctx, key, bundlePath, false, "ledger bundle ("+sc+")"); err != nil { + return nil, 0, err + } + bundlePaths = append(bundlePaths, bundlePath) + } + return bundlePaths, goldenFetchSecs, nil +} diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/runner/main.go b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/runner/main.go new file mode 100644 index 000000000..b9009d3e3 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/runner/main.go @@ -0,0 +1,8 @@ +// Command runner runs the apply-load ingestion benchmark on the box. +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/load-test/testdata/apply-load-v27-oz.cfg b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/testdata/apply-load-v27-oz.cfg similarity index 100% rename from cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/testdata/apply-load-v27-oz.cfg rename to cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/testdata/apply-load-v27-oz.cfg diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/testdata/apply-load-v27-sac.cfg b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/testdata/apply-load-v27-sac.cfg similarity index 100% rename from cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/testdata/apply-load-v27-sac.cfg rename to cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/testdata/apply-load-v27-sac.cfg diff --git a/cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/testdata/apply-load-v27-soroswap.cfg b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/testdata/apply-load-v27-soroswap.cfg similarity index 100% rename from cmd/stellar-rpc/internal/integrationtest/infrastructure/load-test/testdata/apply-load-v27-soroswap.cfg rename to cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/ingest-load-test/testdata/apply-load-v27-soroswap.cfg