diff --git a/.github/workflows/ec2-leg.yml b/.github/workflows/ec2-leg.yml index f3cbb21d7..fa3851713 100644 --- a/.github/workflows/ec2-leg.yml +++ b/.github/workflows/ec2-leg.yml @@ -210,8 +210,8 @@ jobs: 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. + # Polls S3 for the result object and relays found/passed as step outputs + # for the fail gate below. # 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 diff --git a/.github/workflows/load-test-coordinator.yml b/.github/workflows/load-test-coordinator.yml index 5f66aaa50..1610f76f2 100644 --- a/.github/workflows/load-test-coordinator.yml +++ b/.github/workflows/load-test-coordinator.yml @@ -19,6 +19,11 @@ on: required: false type: string default: '' + pr_number: + description: 'PR to report results to (default: inferred from the tested branch)' + 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. @@ -50,14 +55,21 @@ jobs: env: GH_TOKEN: ${{ github.token }} REF: ${{ inputs.target_ref || github.ref_name }} + PR: ${{ inputs.pr_number }} 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)" + if [ -z "$PR" ]; then + PR="$(gh pr list --repo "${{ github.repository }}" --state open \ + --head "$REF" --json number --jq '.[0].number // ""' 2>/dev/null || true)" + fi + 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","budget":225}, + {"label":"Backfill ingestion","run_label":"backfill","script":"cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh","budget":345} + ]' { 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"}]' + echo "legs=$(jq -c . <<<"$LEGS")" } >> "$GITHUB_OUTPUT" # The fan-out: one matrix entry per leg. @@ -74,6 +86,7 @@ jobs: target_ref: ${{ needs.plan.outputs.target_sha }} run_label: ${{ matrix.leg.run_label }} leg_script: ${{ matrix.leg.script }} + budget_minutes: ${{ matrix.leg.budget }} report: name: Aggregate + report 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 new file mode 100644 index 000000000..e43a5b766 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/run-backfill.sh @@ -0,0 +1,9 @@ +# 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. +LEG_TITLE="Backfill ingestion" + +bootstrap_box +run_leg ./cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner 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 new file mode 100644 index 000000000..0c71efae4 --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/instantiate.go @@ -0,0 +1,223 @@ +package main + +import ( + "bufio" + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "regexp" + "strconv" + "time" + + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/s3" + + "github.com/stellar/go-stellar-sdk/ingest/ledgerbackend" + + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/harness" +) + +const ( + // runner runs w/ cwd = repo root, so paths are relative to there + legDir = "cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test" + corePath = "/usr/local/bin/stellar-core" // fetched from S3 +) + +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. +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") + resultsFile = harness.Env("RESULTS_FILE", "/tmp/results.md") + resultKey = os.Getenv("RESULT_KEY") + targetSHA = os.Getenv("TARGET_SHA") + runID = harness.Env("RUN_ID", "manual") + // ~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(workDir, "stellar-rpc-bin") // built here (the repo checkout is in WORK_DIR) + ) + repoRoot, err := os.Getwd() + if err != nil { + return err + } + bail := func(format string, args ...any) error { + return harness.BailInstance(resultsFile, "Backfill ingestion", runID, 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(region)) + if err != nil { + return bail("loading AWS config: %v", err) + } + fetch := &harness.S3Fetcher{Client: s3.NewFromConfig(awsCfg), Bucket: bucket} + + if err := fetch.FetchVerified(ctx, "core/stellar-core.zst", corePath, true, "stellar-core"); 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(workDir, "captive-core-pubnet.cfg") + if err := os.WriteFile(coreCfg, ledgerbackend.PubnetDefaultConfig, 0o644); err != nil { + return bail("writing captive-core config: %v", err) + } + + cfgPath, err := renderConfig(repoRoot, workDir, coreCfg, retention) + 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) + if err != nil { + return bail("%v", err) + } + ingested := hi - lo + 1 + if ingested+ledgerThreshold < want { + 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(targetSHA, retention, lo, hi, ingested, elapsed) + if err := os.WriteFile(resultsFile, []byte(md), 0o644); err != nil { + return bail("writing results: %v", err) + } + if err := harness.PublishResult( + ctx, fetch.Client, bucket, resultKey, "ok", runID, targetSHA, resultsFile, ""); err != nil { + return bail("publishing result: %v", err) + } + 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) { + tmpl, err := os.ReadFile(filepath.Join(repoRoot, legDir, "testdata", "backfill-pubnet.toml.tmpl")) + if err != nil { + return "", err + } + mapping := func(in string) string { + switch in { + case "CAPTIVE_CORE_CONFIG_PATH": + return coreCfg + case "CAPTIVE_CORE_STORAGE_PATH": + return filepath.Join(workDir, "core-storage") + case "DB_PATH": + return filepath.Join(workDir, "backfill.sqlite") + case "STELLAR_CORE_BINARY_PATH": + return corePath + case "HISTORY_RETENTION_WINDOW": + return retention + default: + return "${" + in + "}" // leave unknown placeholders intact + } + } + body := os.Expand(string(tmpl), mapping) + cfgPath := filepath.Join(workDir, "backfill-rpc.toml") + if err := os.WriteFile(cfgPath, []byte(body), 0o644); err != nil { + return "", err + } + return cfgPath, nil +} + +// 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() + + 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 + } + cmd.Stdout, cmd.Stderr = pw, pw + + start := time.Now() + if err := cmd.Start(); err != nil { + pw.Close() + pr.Close() + return 0, 0, 0, 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 + scanner := bufio.NewScanner(pr) + scanner.Buffer(make([]byte, 64*1024), 1024*1024) + for scanner.Scan() { + line := scanner.Text() + fmt.Fprintln(os.Stderr, line) // tee to the box user-data log (SSM debug tail) + if m := backfillDoneRe.FindStringSubmatch(line); m != nil { + elapsed = time.Since(start) + lo, _ = strconv.Atoi(m[1]) + hi, _ = strconv.Atoi(m[2]) + cancel() // stop the daemon before it starts live ingestion + break + } + } + scanErr := scanner.Err() + if scanErr != nil { + cancel() + } + _ = 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) + } + return 0, 0, 0, fmt.Errorf("daemon exited or hit the %s deadline before backfill completed", deadline) + } + return elapsed, lo, hi, nil +} + +func renderMarkdown(sha, retention string, lo, hi, ingested int, elapsed time.Duration) string { + shortSHA := sha + if len(shortSHA) > 12 { + shortSHA = shortSHA[:12] + } + lps := 0.0 + if s := elapsed.Seconds(); s > 0 { + lps = float64(ingested) / s + } + return fmt.Sprintf("### ⏳ Backfill ingestion — `%s`\n\n"+ + "| Metric | Value |\n|---|---|\n"+ + "| Ledgers ingested | %d (`[%d -> %d]`) |\n"+ + "| Retention window | %s |\n"+ + "| Wall-clock | %s |\n"+ + "| Ledgers/sec | %.1f |\n", + shortSHA, ingested, lo, hi, retention, elapsed.Round(time.Second), lps) +} 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 new file mode 100644 index 000000000..5f2acfa2a --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/runner/main.go @@ -0,0 +1,8 @@ +// Command runner runs the backfill 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/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 new file mode 100644 index 000000000..80927c9fc --- /dev/null +++ b/cmd/stellar-rpc/internal/integrationtest/infrastructure/perf-eval/backfill-test/testdata/backfill-pubnet.toml.tmpl @@ -0,0 +1,36 @@ +# To fill in and use by the backfill leg's runner (instantiate.go renderConfig), +# which substitutes the placeholders below via os.Expand before the daemon reads +# it -- not a standalone config. + +NETWORK_PASSPHRASE = "Public Global Stellar Network ; September 2015" +HISTORY_ARCHIVE_URLS = ["https://history.stellar.org/prd/core-live/core_live_001"] +ENDPOINT = "localhost:8000" +LOG_LEVEL = "info" + +# Captive core is constructed at startup (so the binary + config must exist) +# even though backfill itself reads from the datastore below. +CAPTIVE_CORE_CONFIG_PATH = "${CAPTIVE_CORE_CONFIG_PATH}" +STELLAR_CORE_BINARY_PATH = "${STELLAR_CORE_BINARY_PATH}" +CAPTIVE_CORE_STORAGE_PATH = "${CAPTIVE_CORE_STORAGE_PATH}" +CHECKPOINT_FREQUENCY = 64 + +DB_PATH = "${DB_PATH}" + +# Backfill the retention window from the datastore on startup, then stop (the +# leg kills the daemon at the completion log line). +BACKFILL = true +SERVE_LEDGERS_FROM_DATASTORE = true +HISTORY_RETENTION_WINDOW = ${HISTORY_RETENTION_WINDOW} +INGESTION_TIMEOUT = "50m0s" + +[datastore_config] + type = "S3" + + [datastore_config.params] + destination_bucket_path = "aws-public-blockchain/v1.1/stellar/ledgers/pubnet" + region = "us-east-2" + + [datastore_config.schema] + FileExtension = "zst" + files_per_partition = 64000 + ledgers_per_file = 1