ci: scrape WAL metrics on failed loads and keep logic out of the benchmark YAML

Scrape the object store WAL metrics before stopping a cell's server whether the load succeeded or failed, so failed cells keep their stall, poison and create failure counters. Build the summary from the plan so that cells which never ran are listed and leave their ratios undefined. Move budget, region, runtime image, manifest and credential scan logic from the workflow into the helper scripts, and drive the cell lifecycle in the tests through fake docker and curl binaries. Signed-off-by: jeremyhi <fengjiachun@gmail.com>
2026-10-12 23:12:28 +00:00 · 2026-10-12 17:12:12 +08:00
parent 774a4b9432
commit f34b62916f
5 changed files with 314 additions and 103 deletions
@@ -1,14 +1,40 @@
 #!/usr/bin/env bash
 # Host side of the object store WAL S3 benchmark (oswal-s3-bench.yml).
-#   wal-disk DIR  mount the instance store at DIR when the host has one
-#   generate      generate the dataset once with the o11ybench runtime
-#   cells         run every cell of $PLAN, each on a fresh GreptimeDB container
+#   check-runtime IMAGE  resolve the o11ybench runtime and check it carries the loader and generator
+#   pull                 pull the runtime and GreptimeDB images and record their metadata
+#   wal-disk DIR         mount the instance store at DIR when the host has one
+#   generate             generate the dataset once with the o11ybench runtime
+#   cells                run every cell of $PLAN, each on a fresh GreptimeDB container
+#   scrub                refuse results that contain a credential value
 set -euo pipefail

-: "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}"
+here=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+
 # Same isolation as the o11ybench runtime's own docker run.
-runtime=(docker run --rm --user "$(id -u):$(id -g)" --label "o11ybench.owner=$OWNER"
-    --security-opt no-new-privileges --cap-drop ALL)
+runtime() {
+    docker run --rm --user "$(id -u):$(id -g)" --label "o11ybench.owner=$OWNER" \
+        --security-opt no-new-privileges --cap-drop ALL "$@"
+}
+
+check_runtime() {
+    local image=$1 resolved
+    [[ "$image" == public.ecr.aws/* ]] || { echo 'The runtime image must come from AWS ECR Public' >&2; return 1; }
+    bash "$here/pull-benchmark-image.sh" "$image"
+    resolved=$(docker image inspect "$image" | jq -er '.[0].RepoDigests[0]')
+    docker run --rm --entrypoint sh "$resolved" -c \
+        'test -f scripts/agent-observability/greptimedb.sh && test -f scripts/generate_agent_observations.py' \
+        || { echo 'The runtime image lacks the agent observability loader or generator' >&2; return 1; }
+    printf 'runtime_image=%s\n' "$resolved" >> "${GITHUB_OUTPUT:-/dev/stdout}"
+}
+
+pull_images() {
+    : "${RESULTS:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_TAG:?}"
+    mkdir -p "$RESULTS/images"
+    bash "$here/pull-benchmark-image.sh" "$RUNTIME_IMAGE"
+    bash "$here/pull-benchmark-image.sh" "greptime/greptimedb:$GREPTIMEDB_TAG"
+    docker image inspect "$RUNTIME_IMAGE" > "$RESULTS/images/runtime.json"
+    docker image inspect "greptime/greptimedb:$GREPTIMEDB_TAG" > "$RESULTS/images/greptimedb.json"
+}

 wal_disk() {
    local root=$1 device
@@ -24,11 +50,11 @@ wal_disk() {
 }

 generate() {
-    local args
-    mapfile -t args < <(jq -r '.generator_args[]' "$PLAN")
+    : "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}"
+    local args=() arg
+    while IFS= read -r arg; do args+=("$arg"); done < <(jq -r '.generator_args[]' "$PLAN")
    mkdir -p "$CORPUS_DIR"
-    "${runtime[@]}" --name "$OWNER-generator" \
-        --mount "type=bind,src=$CORPUS_DIR,dst=/work/corpus" \
+    runtime --name "$OWNER-generator" --mount "type=bind,src=$CORPUS_DIR,dst=/work/corpus" \
        "$RUNTIME_IMAGE" scripts/generate_agent_observations.py "${args[@]}" \
        --output /work/corpus/agent_observations.jsonl \
        --summary-output /work/corpus/summary.json --evidence-output /work/corpus/evidence.json \
@@ -36,22 +62,50 @@ generate() {
    jq -e --argjson rows "$(jq .rows "$PLAN")" '.line_count == $rows' "$CORPUS_DIR/summary.json" >/dev/null
 }

-# One cell in its own process, so errexit holds inside it.
+# Scrape once, after the load succeeded or failed, while the server still runs.
+scrape_metrics() {
+    if curl --noproxy '*' -fsS --max-time 30 http://127.0.0.1:4000/metrics > "$dir/metrics.all"; then
+        grep -E '^(# (HELP|TYPE) )?greptime_logstore_object_store_wal_' "$dir/metrics.all" > "$dir/metrics.prom" || true
+    fi
+    rm -f "$dir/metrics.all"
+}
+
+finish() {
+    local rc=$?
+    trap - EXIT
+    set +e
+    if [[ -n "$cid" ]]; then
+        scrape_metrics
+        docker stop -t 120 "$cid" >/dev/null
+        docker logs "$cid" > "$dir/server.log" 2>&1
+        docker rm -fv "$cid" >/dev/null
+    fi
+    [[ -z "$wal_dir" ]] || sudo -n rm -rf "$wal_dir"
+    jq -n --argjson spec "$spec" --arg device "$wal_device" --arg version "$version" \
+        --arg load_seconds "$load_seconds" --argjson exit_code "$rc" \
+        '$spec + {wal_device: $device, version: $version, exit_code: $exit_code,
+          load_wall_seconds: ($load_seconds | tonumber? // null)}' > "$dir/cell.json"
+    exit "$rc"
+}
+
+# One cell in its own process, so errexit holds inside it. The cell's state is
+# global because the EXIT trap runs after this function returns.
 cell() {
-    : "${GREPTIMEDB_IMAGE:?}" "${RUN_PREFIX:?}" "${WAL_ROOT:?}" "${S3_BUCKET:?}" "${S3_REGION:?}"
+    : "${PLAN:?}" "${RESULTS:?}" "${CORPUS_DIR:?}" "${OWNER:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_IMAGE:?}"
+    : "${RUN_PREFIX:?}" "${WAL_ROOT:?}" "${S3_BUCKET:?}" "${S3_REGION:?}"
    : "${AWS_ACCESS_KEY_ID:?}" "${AWS_SECRET_ACCESS_KEY:?}"
-    # Globals, not locals: the EXIT trap runs after this function returns.
    spec=$1 cid= wal_dir= version= load_seconds=
-    local concurrency config name
-    id=$(jq -r .id <<< "$spec"); concurrency=$(jq -r .concurrency <<< "$spec"); config=$(jq -r .config <<< "$spec")
+    local id concurrency name setting started ended
+    id=$(jq -r .id <<< "$spec"); concurrency=$(jq -r .concurrency <<< "$spec")
    dir="$RESULTS/cells/$id"; name="$OWNER-$id"
    mkdir -p "$dir"
-    # The prefix holds data/ and the WAL's default prefix wal/ side by side.
+    # The root holds data/ and the WAL's default prefix wal/ side by side.
    local settings=(STORAGE__TYPE=S3 "STORAGE__BUCKET=$S3_BUCKET" "STORAGE__REGION=$S3_REGION"
        "STORAGE__ROOT=$RUN_PREFIX/$id")
    [[ -z "${S3_CONTAINER_ENDPOINT:-}" ]] || settings+=("STORAGE__ENDPOINT=$S3_CONTAINER_ENDPOINT")
-    mapfile -t -O "${#settings[@]}" settings < <(jq -r '.wal | to_entries[] | "WAL__\(.key)=\(.value)"' <<< "$spec")
-    local docker_env=() mounts=() setting
+    while IFS= read -r setting; do settings+=("$setting"); done \
+        < <(jq -r '.wal | to_entries[] | "WAL__\(.key)=\(.value)"' <<< "$spec")
+    local docker_env=() mounts=()
    for setting in "${settings[@]}"; do
        docker_env+=(-e "GREPTIMEDB_STANDALONE__$setting")
        printf 'GREPTIMEDB_STANDALONE__%s\n' "$setting" >> "$dir/config.env"
@@ -68,26 +122,10 @@ cell() {
    else
        wal_device="s3://$S3_BUCKET/$RUN_PREFIX/$id/wal"
    fi
-    finish() {
-        local rc=$?
-        trap - EXIT
-        set +e
-        if [[ -n "$cid" ]]; then
-            docker stop -t 120 "$cid" >/dev/null
-            docker logs "$cid" > "$dir/server.log" 2>&1
-            docker rm -fv "$cid" >/dev/null
-        fi
-        [[ -z "$wal_dir" ]] || sudo -n rm -rf "$wal_dir"
-        jq -n --argjson spec "$spec" --arg device "$wal_device" --arg version "$version" \
-            --arg load_seconds "$load_seconds" --argjson exit_code "$rc" \
-            '$spec + {wal_device: $device, version: $version, exit_code: $exit_code,
-              load_wall_seconds: ($load_seconds | tonumber? // null)}' > "$dir/cell.json"
-        exit "$rc"
-    }
    trap finish EXIT
    cid=$(docker run -d --init --name "$name" --label "o11ybench.owner=$OWNER" \
        --add-host host.docker.internal:host-gateway \
-        -p 127.0.0.1:4000:4000 -p 127.0.0.1:4001:4001 "${docker_env[@]}" "${mounts[@]}" \
+        -p 127.0.0.1:4000:4000 -p 127.0.0.1:4001:4001 "${docker_env[@]}" ${mounts[@]+"${mounts[@]}"} \
        "$GREPTIMEDB_IMAGE" standalone start --http-addr=0.0.0.0:4000 --grpc-bind-addr=0.0.0.0:4001)
    local deadline=$((SECONDS + 300))
    until curl --noproxy '*' -fsS --max-time 5 http://127.0.0.1:4000/health >/dev/null 2>&1; do
@@ -97,10 +135,9 @@ cell() {
    done
    version=$(docker exec "$cid" greptime --version | sed -n 's/^version: //p')
    [[ -n "$version" ]]
-    local started ended
    started=$(date +%s.%N)
    # Only the load phase, run exactly as run-agent-observability.sh runs it.
-    "${runtime[@]}" --name "$name-load" --network host \
+    runtime --name "$name-load" --network host \
        --mount "type=bind,src=$CORPUS_DIR,dst=/work/corpus,readonly" --mount "type=bind,src=$dir,dst=/results" \
        -e GRPC_ENDPOINT=127.0.0.1:4001 -e "DATA_PROFILE=$(jq -r .data_profile "$PLAN")" -e RESULT_NAME=greptimedb-1 \
        -e "ENDPOINT=http://127.0.0.1:4000/v1/sql?db=public&format=greptimedb_v1" \
@@ -112,23 +149,34 @@ cell() {
    load_seconds=$(awk -v a="$started" -v b="$ended" 'BEGIN {printf "%.3f", b - a}')
    jq -e --argjson rows "$(jq .rows "$PLAN")" '.count_before == 0 and .count_after == $rows' \
        "$dir/greptimedb-1/load-result.json" >/dev/null
-    curl --noproxy '*' -fsS --max-time 30 http://127.0.0.1:4000/metrics > "$dir/metrics.all"
-    grep -E '^(# (HELP|TYPE) )?greptime_logstore_object_store_wal_' "$dir/metrics.all" > "$dir/metrics.prom" || true
-    rm "$dir/metrics.all"
+}
+
+scrub() {
+    : "${RESULTS:?}"
+    [[ -n "${AWS_ACCESS_KEY_ID:-}" && -n "${AWS_SECRET_ACCESS_KEY:-}" && -d "$RESULTS" ]] || return 0
+    if grep -rqF -e "$AWS_ACCESS_KEY_ID" -e "$AWS_SECRET_ACCESS_KEY" "$RESULTS"; then
+        echo '::error::A credential value appears in the results; removing them instead of uploading' >&2
+        rm -rf "$RESULTS"
+        return 1
+    fi
 }

 case "${1:-}" in
+    check-runtime) check_runtime "${2:?}" ;;
+    pull) pull_images ;;
    wal-disk) wal_disk "${2:?}" ;;
    generate) generate ;;
    cell) cell "${2:?}" ;;
    cells)
-        mapfile -t specs < <(jq -c '.cells[]' "$PLAN")
+        : "${PLAN:?}"
        failed=0
-        for spec in "${specs[@]}"; do
-            echo "::group::cell $(jq -r .id <<< "$spec")"
-            bash "${BASH_SOURCE[0]}" cell "$spec" || { echo "::error::cell $(jq -r .id <<< "$spec") failed"; failed=1; }
+        while IFS= read -r spec; do
+            id=$(jq -r .id <<< "$spec")
+            echo "::group::cell $id"
+            bash "${BASH_SOURCE[0]}" cell "$spec" < /dev/null || { echo "::error::cell $id failed"; failed=1; }
            echo '::endgroup::'
-        done
+        done < <(jq -c '.cells[]' "$PLAN")
        exit "$failed" ;;
-    *) echo "usage: $0 wal-disk DIR | generate | cells" >&2; exit 2 ;;
+    scrub) scrub ;;
+    *) echo "usage: $0 check-runtime IMAGE | pull | wal-disk DIR | generate | cells | scrub" >&2; exit 2 ;;
 esac
@@ -87,6 +87,14 @@ def load_cell(path: Path) -> dict:
    return row


+def missing_cell(spec: dict) -> dict:
+    """A planned cell that never recorded a result, so no ratio can use its round."""
+    row = {key: None for key in ("exit_code", "wal_device", "version", "load_wall_seconds", "rows",
+                                 "load_seconds", "rows_per_second", "append_ack_p50_seconds",
+                                 "append_ack_p99_seconds", "creates_per_minute", *COUNTERS)}
+    return row | {key: spec[key] for key in ("id", "round", "concurrency", "config")} | {"missing": True}
+
+
 def ratios(rows: list) -> list:
    """D/R and E/R per round and their geometric mean over the rounds, per concurrency."""
    speed = {(r["round"], r["concurrency"], r["config"]): r["rows_per_second"] for r in rows}
@@ -110,18 +118,21 @@ def fmt(value, digits=0, scale=1.0) -> str:

 def render(run: dict, rows: list, pairs: list) -> str:
    lines = [f"## Object store WAL against S3 ({'smoke, MinIO' if run.get('smoke') else run.get('instance_type')})", ""]
-    lines += [f"GreptimeDB `{run.get('greptimedb_image')}` ({rows[0]['version'] if rows else '—'}); "
+    lines += [f"GreptimeDB `{run.get('greptimedb_image')}` ({next((r['version'] for r in rows if r['version']), '—')}); "
              f"{fmt(run.get('rows'))} rows in {run.get('batch_rows')}-row batches; bucket region {run.get('region')}.", ""]
    lines += ["| Cell | Rows/s | Load s | Wall s | Rows | Ack p50 ms | Ack p99 ms | Creates | Creates/min "
              "| Stalled | Poisoned | Create failures | Conflicts | WAL on |",
              "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |"]
    for r in rows:
-        cell = r["id"] if r["exit_code"] == 0 else f"{r['id']} (failed, exit {r['exit_code']})"
+        if r.get("missing"):
+            cell = f"{r['id']} (not run)"
+        else:
+            cell = r["id"] if r["exit_code"] == 0 else f"{r['id']} (failed, exit {r['exit_code']})"
        lines.append(" | ".join([
            f"| {cell}", fmt(r["rows_per_second"]), fmt(r["load_seconds"], 1), fmt(r["load_wall_seconds"], 1),
            fmt(r["rows"]), fmt(r["append_ack_p50_seconds"], 1, 1000), fmt(r["append_ack_p99_seconds"], 1, 1000),
            *(fmt(r[c]) for c in COUNTERS[:1]), fmt(r["creates_per_minute"], 1),
-            *(fmt(r[c]) for c in COUNTERS[1:]), f"`{r['wal_device']}` |"]))
+            *(fmt(r[c]) for c in COUNTERS[1:]), f"`{r['wal_device'] or '—'}` |"]))
    lines += ["", "| Concurrency | Ratio | Per round | Geometric mean |", "| ---: | --- | --- | ---: |"]
    for p in pairs:
        lines.append(f"| {p['concurrency']} | {p['ratio']} | {', '.join(fmt(v, 3) for v in p['per_round'])} "
@@ -137,11 +148,16 @@ def main() -> None:
    run_path = args.root / "run.json"
    run = json.loads(run_path.read_text()) if run_path.exists() else {}
    plan_path = args.root / "plan.json"
-    order = [cell["id"] for cell in json.loads(plan_path.read_text())["cells"]] if plan_path.exists() else []
+    planned = json.loads(plan_path.read_text())["cells"] if plan_path.exists() else []
    cells = {path.name: load_cell(path) for path in (args.root / "cells").glob("*") if (path / "cell.json").exists()}
-    rows = [cells[i] for i in order if i in cells] + [cells[i] for i in sorted(cells) if i not in order]
+    # The plan, not what happened to run, defines the expected cells and rounds.
+    rows = [cells.get(spec["id"]) or missing_cell(spec) for spec in planned]
+    rows += [cells[i] for i in sorted(cells) if i not in {spec["id"] for spec in planned}]
    pairs = ratios(rows)
-    (args.root / "summary.json").write_text(json.dumps({"run": run, "cells": rows, "ratios": pairs}, indent=2) + "\n")
+    missing = [r["id"] for r in rows if r.get("missing")]
+    (args.root / "summary.json").write_text(json.dumps(
+        {"run": run, "planned_cells": len(planned), "missing_cells": missing, "cells": rows, "ratios": pairs},
+        indent=2) + "\n")
    print(render(run, rows, pairs) if rows else "No benchmark cell completed.\n")


@@ -18,10 +18,10 @@
 # dependencies = ["boto3>=1.35,<2"]
 # ///

-"""Plan, image tag and bucket helpers of the object store WAL S3 benchmark.
+"""Plan, image tag, run manifest and bucket helpers of the object store WAL S3 benchmark.

-`plan` and `tag` need only the standard library; `probe` and `cleanup` need
-boto3 and run under `uv run`.
+`plan`, `tag` and `manifest` need only the standard library; `probe` and
+`cleanup` need boto3 and run under `uv run`.
 """

 from __future__ import annotations
@@ -33,6 +33,7 @@ import re
 import sys
 import time
 import urllib.request
+from pathlib import Path

 SMOKE_ROWS = 50_000
 # The o11ybench data profiles; a matching row count generates the canonical dataset.
@@ -46,10 +47,11 @@ WAL_CONFIGS = {
 ORDERS = (("R", "D", "E"), ("E", "D", "R"))
 NIGHTLY = re.compile(r"v\d+\.\d+\.\d+(?:-[0-9A-Za-z.]+)?-nightly-(\d{8})")
 TAG = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}")
+DIGEST = re.compile(r".+@sha256:[0-9a-f]{64}")


-def append_output(name: str, value: str) -> None:
-    if path := os.environ.get("GITHUB_OUTPUT"):
+def append_github(variable: str, name: str, value: str) -> None:
+    if path := os.environ.get(variable):
        with open(path, "a") as output:
            output.write(f"{name}={value}\n")

@@ -60,6 +62,30 @@ def positive(name: str, value: str, upper: int) -> int:
    return int(value)


+def validate_budgets(timeout_minutes: str, ttl_hours: str) -> None:
+    timeout, ttl = positive("timeout", timeout_minutes, 360), positive("ttl", ttl_hours, 168)
+    # Provisioning, teardown and queueing need two hours beyond the benchmark job.
+    if ttl * 60 < timeout + 120:
+        raise ValueError(f"janitor TTL {ttl}h must cover the {timeout}-minute timeout plus 2 hours")
+
+
+def check_regions(bucket: str, runner: str) -> None:
+    if not bucket or bucket != runner:
+        raise ValueError(f"bucket region {bucket!r} differs from runner region {runner!r}; "
+                         "a cross-region bucket measures the WAN, not S3")
+
+
+def build_manifest(plan: dict, greptimedb_inspect: list, runtime_image: str,
+                   instance_type: str, region: str, prefix: str) -> dict:
+    image = next((d for d in greptimedb_inspect[0].get("RepoDigests") or [] if DIGEST.fullmatch(d)), None)
+    if image is None:
+        raise ValueError("the GreptimeDB image has no sha256 repository digest")
+    return {"smoke": plan["smoke"], "rows": plan["rows"], "rounds": plan["rounds"],
+            "concurrency": plan["concurrency"], "batch_rows": plan["batch_rows"],
+            "greptimedb_image": image, "runtime_image": runtime_image,
+            "instance_type": instance_type, "region": region, "prefix": prefix}
+
+
 def build_plan(concurrency: str, rows: str, batch_rows: str, rounds: str, smoke: bool) -> dict:
    levels = [positive("concurrency", level.strip(), 256) for level in concurrency.split(",")]
    if len(set(levels)) != len(levels):
@@ -149,8 +175,16 @@ def main() -> int:
    for name in ("concurrency", "rows", "batch-rows", "rounds"):
        plan.add_argument(f"--{name}", required=True)
    plan.add_argument("--smoke", choices=("true", "false"), required=True)
+    plan.add_argument("--timeout-minutes")
+    plan.add_argument("--ttl-hours")
    plan.add_argument("--output")
    commands.add_parser("tag").add_argument("requested")
+    regions = commands.add_parser("regions")
+    regions.add_argument("--bucket", required=True)
+    regions.add_argument("--runner", required=True)
+    manifest = commands.add_parser("manifest")
+    for name in ("results", "runtime-image", "instance-type", "region", "prefix"):
+        manifest.add_argument(f"--{name}", required=True)
    for name in ("probe", "cleanup"):
        bucket = commands.add_parser(name)
        bucket.add_argument("--bucket", required=True)
@@ -159,6 +193,8 @@ def main() -> int:
        bucket.add_argument("--prefix", required=True)
    args = parser.parse_args()
    if args.command == "plan":
+        if args.timeout_minutes or args.ttl_hours:
+            validate_budgets(args.timeout_minutes or "", args.ttl_hours or "")
        result = build_plan(args.concurrency, args.rows, args.batch_rows, args.rounds, args.smoke == "true")
        if args.output:
            with open(args.output, "w") as output:
@@ -166,8 +202,19 @@ def main() -> int:
        print(f"{len(result['cells'])} cells, {result['rows']} rows, profile {result['data_profile']}")
    elif args.command == "tag":
        tag = resolve_tag(args.requested)
-        append_output("greptimedb_tag", tag)
+        append_github("GITHUB_OUTPUT", "greptimedb_tag", tag)
        print(f"GreptimeDB image tag: {tag}")
+    elif args.command == "regions":
+        check_regions(args.bucket, args.runner)
+        print(f"bucket and runner both in {args.runner}")
+    elif args.command == "manifest":
+        results = Path(args.results)
+        manifest = build_manifest(json.loads((results / "plan.json").read_text()),
+                                  json.loads((results / "images" / "greptimedb.json").read_text()),
+                                  args.runtime_image, args.instance_type, args.region, args.prefix)
+        (results / "run.json").write_text(json.dumps(manifest, indent=2) + "\n")
+        append_github("GITHUB_ENV", "GREPTIMEDB_IMAGE", manifest["greptimedb_image"])
+        print(f"GreptimeDB image: {manifest['greptimedb_image']}")
    else:
        if not re.fullmatch(r"oswal-bench/[0-9]+-[0-9]+", args.prefix):
            raise ValueError(f"refusing a prefix outside oswal-bench/<run>: {args.prefix!r}")
@@ -176,4 +223,8 @@ def main() -> int:


 if __name__ == "__main__":
-    sys.exit(main())
+    try:
+        sys.exit(main())
+    except ValueError as error:
+        print(f"::error::{error}", file=sys.stderr)
+        sys.exit(1)
@@ -82,11 +82,12 @@ jobs:
      run: |
        python3 greptimedb/tests/perf/test_oswal_s3_bench_scripts.py
        python3 greptimedb/tests/perf/test_ci_runner_scripts.py
-    - name: Validate plan and image tag
+    - name: Validate plan, budgets and image tag
      id: plan
      run: |
        python3 greptimedb/.github/scripts/oswal-s3-bench.py plan --smoke "$SMOKE" \
-          --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS"
+          --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS" \
+          --timeout-minutes "$BENCHMARK_TIMEOUT_MINUTES" --ttl-hours "$JANITOR_TTL_HOURS"
        python3 greptimedb/.github/scripts/oswal-s3-bench.py tag "$GREPTIMEDB_TAG"
    - name: Resolve CI runner
      id: runner
@@ -95,27 +96,10 @@ jobs:
      env:
        BUCKET_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
        RUNNER_REGION: ${{ vars.EC2_RUNNER_REGION }}
-      run: |
-        [[ -n "$BUCKET_REGION" && "$BUCKET_REGION" == "$RUNNER_REGION" ]] || {
-          echo "::error::Bucket region '$BUCKET_REGION' differs from runner region '$RUNNER_REGION'; a cross-region bucket measures the WAN, not S3"
-          exit 1
-        }
-    - name: Validate budgets and runtime image
+      run: python3 greptimedb/.github/scripts/oswal-s3-bench.py regions --bucket "$BUCKET_REGION" --runner "$RUNNER_REGION"
+    - name: Check the runtime image
      id: runtime
-      run: |
-        set -euo pipefail
-        for value in "$BENCHMARK_TIMEOUT_MINUTES" "$JANITOR_TTL_HOURS"; do
-          [[ "$value" =~ ^[1-9][0-9]{0,3}$ ]] || { echo 'Budgets must be positive integers'; exit 1; }
-        done
-        ((BENCHMARK_TIMEOUT_MINUTES <= 360 && JANITOR_TTL_HOURS <= 168))
-        ((JANITOR_TTL_HOURS * 60 >= BENCHMARK_TIMEOUT_MINUTES + 120)) || { echo 'Janitor TTL must cover the timeout plus 2 hours'; exit 1; }
-        [[ "$RUNTIME_IMAGE" == public.ecr.aws/* ]]
-        bash greptimedb/.github/scripts/pull-benchmark-image.sh "$RUNTIME_IMAGE"
-        resolved=$(docker image inspect "$RUNTIME_IMAGE" | jq -er '.[0].RepoDigests[0]')
-        docker run --rm --entrypoint sh "$resolved" -c \
-          'test -f scripts/agent-observability/greptimedb.sh && test -f scripts/generate_agent_observations.py' \
-          || { echo 'Runtime image lacks the agent observability loader or generator'; exit 1; }
-        printf 'runtime_image=%s\n' "$resolved" >> "$GITHUB_OUTPUT"
+      run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh check-runtime "$RUNTIME_IMAGE"
  provision:
    needs: validate
    if: ${{ !inputs.smoke }}
@@ -185,20 +169,11 @@ jobs:
        docker compose -f greptimedb/tests-integration/fixtures/docker-compose.yml up -d --wait minio
    - name: Pull images and write the plan
      run: |
-        set -euo pipefail
-        mkdir -p "$RESULTS/images"
-        bash greptimedb/.github/scripts/pull-benchmark-image.sh "$RUNTIME_IMAGE"
-        bash greptimedb/.github/scripts/pull-benchmark-image.sh "greptime/greptimedb:$GREPTIMEDB_TAG"
-        docker image inspect "$RUNTIME_IMAGE" > "$RESULTS/images/runtime.json"
-        docker image inspect "greptime/greptimedb:$GREPTIMEDB_TAG" > "$RESULTS/images/greptimedb.json"
-        image=$(jq -er '.[0].RepoDigests[0] | select(test("@sha256:[a-f0-9]{64}$"))' "$RESULTS/images/greptimedb.json")
-        printf 'GREPTIMEDB_IMAGE=%s\n' "$image" >> "$GITHUB_ENV"
+        bash greptimedb/.github/scripts/oswal-s3-bench-run.sh pull
        python3 greptimedb/.github/scripts/oswal-s3-bench.py plan --smoke "$SMOKE" --output "$PLAN" \
          --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS"
-        jq -n --slurpfile plan "$PLAN" --arg image "$image" --arg runtime "$RUNTIME_IMAGE" \
-          --arg instance_type "$INSTANCE_TYPE" --arg region "$S3_REGION" --arg prefix "$RUN_PREFIX" \
-          '{smoke: $plan[0].smoke, rows: $plan[0].rows, batch_rows: $plan[0].batch_rows, greptimedb_image: $image,
-            runtime_image: $runtime, instance_type: $instance_type, region: $region, prefix: $prefix}' > "$RESULTS/run.json"
+        python3 greptimedb/.github/scripts/oswal-s3-bench.py manifest --results "$RESULTS" \
+          --runtime-image "$RUNTIME_IMAGE" --instance-type "$INSTANCE_TYPE" --region "$S3_REGION" --prefix "$RUN_PREFIX"
    - name: Probe the bucket
      env:
        AWS_ACCESS_KEY_ID: ${{ inputs.smoke && 'superpower_ci_user' || secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
@@ -231,14 +206,7 @@ jobs:
      env:
        AWS_ACCESS_KEY_ID: ${{ inputs.smoke && 'superpower_ci_user' || secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
        AWS_SECRET_ACCESS_KEY: ${{ inputs.smoke && 'superpower_password' || secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
-      run: |
-        mkdir -p "$RESULTS"
-        [[ -n "$AWS_ACCESS_KEY_ID" && -n "$AWS_SECRET_ACCESS_KEY" ]] || exit 0
-        if grep -rqF -e "$AWS_ACCESS_KEY_ID" -e "$AWS_SECRET_ACCESS_KEY" "$RESULTS"; then
-          echo '::error::A credential value appears in the results; not uploading them'
-          rm -rf "$RESULTS"
-          exit 1
-        fi
+      run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh scrub
    - name: Upload results (no corpus)
      if: always()
      uses: actions/upload-artifact@v4
@@ -17,6 +17,7 @@

 import importlib.util
 import json
+import os
 import subprocess
 import sys
 import tempfile
@@ -86,8 +87,25 @@ class PlanTest(unittest.TestCase):
        with self.assertRaises(ValueError):
            bench.resolve_tag("bad tag")

-    def test_run_script_parses(self):
-        subprocess.run(["bash", "-n", str(SCRIPTS / "oswal-s3-bench-run.sh")], check=True)
+    def test_budgets_and_regions(self):
+        bench.validate_budgets("240", "8")
+        for timeout, ttl in (("360", "7"), ("0", "8"), ("361", "168"), ("240", "169")):
+            with self.subTest(timeout=timeout, ttl=ttl), self.assertRaises(ValueError):
+                bench.validate_budgets(timeout, ttl)
+        bench.check_regions("ap-southeast-1", "ap-southeast-1")
+        for bucket, runner in (("us-east-1", "ap-southeast-1"), ("", "")):
+            with self.subTest(bucket=bucket), self.assertRaisesRegex(ValueError, "cross-region"):
+                bench.check_regions(bucket, runner)
+
+    def test_manifest_takes_the_sha256_digest(self):
+        plan = bench.build_plan("1,8", "2000000", "1000", "2", smoke=False)
+        inspect = [{"RepoDigests": ["greptime/greptimedb@sha256:" + "a" * 64]}]
+        manifest = bench.build_manifest(plan, inspect, "runtime@sha256:" + "b" * 64, "c7id.2xlarge",
+                                        "ap-southeast-1", "oswal-bench/1-1")
+        self.assertEqual(manifest["greptimedb_image"], "greptime/greptimedb@sha256:" + "a" * 64)
+        self.assertEqual((manifest["rows"], manifest["rounds"], manifest["concurrency"]), (2_000_000, 2, [1, 8]))
+        with self.assertRaises(ValueError):
+            bench.build_manifest(plan, [{"RepoDigests": []}], "", "", "", "")


 class SummaryTest(unittest.TestCase):
@@ -138,6 +156,23 @@ class SummaryTest(unittest.TestCase):
            self.assertEqual(d_cell["stalled_appends_total"], 3)
            self.assertIn("| 1-8-D | 80 | 60.0 | 61.0 | 6,000 | 128.0 | 253.4 | 120 | 120.0 | 3 | 0 | — | — |", output)

+    def test_a_planned_round_that_never_ran_leaves_the_ratio_undefined(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            plan = bench.build_plan("8", "6000", "1000", "2", smoke=False)
+            (root / "plan.json").write_text(json.dumps(plan))
+            for cell_id, speed in {"1-8-R": 100.0, "1-8-D": 80.0, "1-8-E": 90.0}.items():
+                self.write_cell(root, cell_id, speed)
+            output = subprocess.run([sys.executable, str(SCRIPTS / "oswal-s3-bench-summary.py"), "--root", str(root)],
+                                    check=True, capture_output=True, text=True).stdout
+            result = json.loads((root / "summary.json").read_text())
+            self.assertEqual(result["missing_cells"], ["2-8-E", "2-8-D", "2-8-R"])
+            self.assertEqual(result["planned_cells"], 6)
+            for ratio in result["ratios"]:
+                self.assertIsNone(ratio["geomean"])
+                self.assertIsNone(ratio["per_round"][1])
+            self.assertIn("2-8-D (not run)", output)
+
    def test_a_failed_cell_leaves_its_ratio_undefined(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
@@ -149,5 +184,98 @@ class SummaryTest(unittest.TestCase):
            self.assertIn("1-1-D (failed, exit 1)", summary.render({}, rows, summary.ratios(rows)))


+FAKE_DOCKER = r"""#!/usr/bin/env bash
+printf '%s\n' "docker $*" >> "$FAKE_LOG"
+case "$1" in
+    run)
+        for arg in "$@"; do [[ "$arg" != -d ]] || { echo "keys=${GREPTIMEDB_STANDALONE__STORAGE__ACCESS_KEY_ID:+set}${GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY:+set}" >> "$FAKE_LOG"; echo fakecid; exit 0; }; done
+        for arg in "$@"; do
+            case "$arg" in type=bind,src=*,dst=/results) results=${arg#type=bind,src=}; results=${results%,dst=/results} ;; esac
+        done
+        [[ "$FAKE_LOAD_RC" == 0 ]] || exit "$FAKE_LOAD_RC"
+        mkdir -p "$results/greptimedb-1"
+        echo "{\"count_before\": 0, \"count_after\": $FAKE_ROWS, \"load_seconds\": 2, \"rows_per_second\": 3000}" \
+            > "$results/greptimedb-1/load-result.json" ;;
+    inspect) echo true ;;
+    exec) echo 'version: 1.4.0-test' ;;
+    logs) echo 'server log' ;;
+esac
+"""
+FAKE_CURL = r"""#!/usr/bin/env bash
+printf '%s\n' "curl $*" >> "$FAKE_LOG"
+for arg in "$@"; do
+    [[ "$arg" != */metrics ]] || { printf '%s\n' 'greptime_logstore_object_store_wal_poisoned_total 1' 'other_total 2'; exit 0; }
+done
+"""
+
+
+class CellLifecycleTest(unittest.TestCase):
+    """Drives the run script against fake docker and curl binaries."""
+
+    def run_cells(self, root, load_rc):
+        bin_dir = root / "bin"
+        bin_dir.mkdir()
+        for name, body in (("docker", FAKE_DOCKER), ("curl", FAKE_CURL)):
+            (bin_dir / name).write_text(body)
+            (bin_dir / name).chmod(0o755)
+        plan = bench.build_plan("8", "6000", "1000", "1", smoke=False)
+        plan["cells"] = [cell for cell in plan["cells"] if cell["config"] != "R"]
+        (root / "results").mkdir()
+        (root / "results" / "plan.json").write_text(json.dumps(plan))
+        env = os.environ | {
+            "PATH": f"{bin_dir}:{os.environ['PATH']}", "FAKE_LOG": str(root / "fake.log"),
+            "FAKE_LOAD_RC": str(load_rc), "FAKE_ROWS": "6000", "PLAN": str(root / "results" / "plan.json"),
+            "RESULTS": str(root / "results"), "CORPUS_DIR": str(root / "corpus"), "OWNER": "o11ybench-oswal-1-1",
+            "RUNTIME_IMAGE": "runtime@sha256:" + "b" * 64, "GREPTIMEDB_IMAGE": "greptimedb@sha256:" + "a" * 64,
+            "RUN_PREFIX": "oswal-bench/1-1", "WAL_ROOT": str(root / "wal"), "S3_BUCKET": "bucket",
+            "S3_REGION": "ap-southeast-1", "AWS_ACCESS_KEY_ID": "AKIAFAKEKEY", "AWS_SECRET_ACCESS_KEY": "fake-secret-value",
+        }
+        script = str(SCRIPTS / "oswal-s3-bench-run.sh")
+        cells = subprocess.run(["bash", script, "cells"], env=env, capture_output=True, text=True, timeout=60)
+        scrub = subprocess.run(["bash", script, "scrub"], env=env, capture_output=True, text=True, timeout=60)
+        return cells, scrub, (root / "fake.log").read_text(), env
+
+    def test_a_failed_load_still_scrapes_metrics_and_removes_the_server(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            cells, scrub, log, _ = self.run_cells(root, load_rc=3)
+            self.assertNotEqual(cells.returncode, 0)
+            self.assertEqual(scrub.returncode, 0, scrub.stderr)
+            for cell_id in ("1-8-D", "1-8-E"):
+                cell = root / "results" / "cells" / cell_id
+                self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], 3)
+                self.assertEqual((cell / "metrics.prom").read_text(),
+                                 "greptime_logstore_object_store_wal_poisoned_total 1\n")
+                self.assertFalse((cell / "metrics.all").exists())
+                self.assertEqual((cell / "server.log").read_text(), "server log\n")
+            self.assertEqual(log.count("docker stop -t 120 fakecid"), 2)
+            self.assertEqual(log.count("docker rm -fv fakecid"), 2)
+
+    def test_a_cell_passes_the_keys_by_name_and_records_a_redacted_config(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            cells, scrub, log, env = self.run_cells(root, load_rc=0)
+            self.assertEqual(cells.returncode, 0, cells.stdout + cells.stderr)
+            self.assertEqual(scrub.returncode, 0, scrub.stderr)
+            self.assertNotIn("fake-secret-value", log)
+            self.assertNotIn("AKIAFAKEKEY", log)
+            self.assertEqual(log.count("keys=setset"), 2)
+            self.assertIn("-e GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY ", log)
+            cell = root / "results" / "cells" / "1-8-D"
+            config = (cell / "config.env").read_text().splitlines()
+            self.assertIn("GREPTIMEDB_STANDALONE__STORAGE__ROOT=oswal-bench/1-1/1-8-D", config)
+            self.assertIn("GREPTIMEDB_STANDALONE__WAL__ACK_MODE=durable", config)
+            self.assertIn("GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY=<redacted>", config)
+            recorded = json.loads((cell / "cell.json").read_text())
+            self.assertEqual((recorded["exit_code"], recorded["version"]), (0, "1.4.0-test"))
+            self.assertEqual(recorded["wal_device"], "s3://bucket/oswal-bench/1-1/1-8-D/wal")
+            # A leaked value makes scrub refuse the results.
+            (cell / "leak.txt").write_text("fake-secret-value")
+            leaked = subprocess.run(["bash", str(SCRIPTS / "oswal-s3-bench-run.sh"), "scrub"], env=env,
+                                    capture_output=True, text=True, timeout=60)
+            self.assertNotEqual(leaked.returncode, 0)
+            self.assertFalse((root / "results").exists())
+
+
 if __name__ == "__main__":
    unittest.main()