diff --git a/.github/scripts/oswal-s3-bench-run.sh b/.github/scripts/oswal-s3-bench-run.sh index 84af2c27032..068f02d584a 100644 --- a/.github/scripts/oswal-s3-bench-run.sh +++ b/.github/scripts/oswal-s3-bench-run.sh @@ -1,11 +1,5 @@ #!/usr/bin/env bash -# Host side of the object store WAL S3 benchmark (oswal-s3-bench.yml). -# 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 +# Host side of oswal-s3-bench.yml: one subcommand per step; each cell of $PLAN gets a fresh GreptimeDB. set -euo pipefail here=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) @@ -64,10 +58,10 @@ generate() { # 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 + curl --noproxy '*' -fsS --max-time 30 http://127.0.0.1:4000/metrics > "$dir/metrics.all" || { rm -f "$dir/metrics.all"; return 1; } + grep -E '^(# (HELP|TYPE) )?greptime_logstore_object_store_wal_' "$dir/metrics.all" > "$dir/metrics.prom" || true rm -f "$dir/metrics.all" + metrics_scraped=true } finish() { @@ -75,15 +69,16 @@ finish() { trap - EXIT set +e if [[ -n "$cid" ]]; then - scrape_metrics + # A load that succeeded without its metrics is not a measurement. + scrape_metrics || { echo "$id: metrics scrape failed" >&2; [[ $rc != 0 ]] || rc=4; } 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, + --arg load_seconds "$load_seconds" --argjson exit_code "$rc" --argjson scraped "$metrics_scraped" \ + '$spec + {wal_device: $device, version: $version, exit_code: $exit_code, metrics_scraped: $scraped, load_wall_seconds: ($load_seconds | tonumber? // null)}' > "$dir/cell.json" exit "$rc" } @@ -92,10 +87,9 @@ finish() { # global because the EXIT trap runs after this function returns. cell() { : "${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:?}" - spec=$1 cid= wal_dir= version= load_seconds= - local id concurrency name setting started ended + : "${RUN_PREFIX:?}" "${WAL_ROOT:?}" "${S3_BUCKET:?}" "${S3_REGION:?}" "${AWS_ACCESS_KEY_ID:?}" "${AWS_SECRET_ACCESS_KEY:?}" + spec=$1 cid= wal_dir= version= load_seconds= metrics_scraped=false id= + local 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" diff --git a/.github/scripts/oswal-s3-bench-summary.py b/.github/scripts/oswal-s3-bench-summary.py index ed1c4197017..67186d422da 100644 --- a/.github/scripts/oswal-s3-bench-summary.py +++ b/.github/scripts/oswal-s3-bench-summary.py @@ -64,8 +64,10 @@ def geomean(values: list) -> float | None: return math.exp(sum(map(math.log, values)) / len(values)) if values and all(v > 0 for v in values) else None -def load_cell(path: Path) -> dict: - cell = json.loads((path / "cell.json").read_text()) +def load_cell(path: Path, spec: dict | None = None) -> dict: + """One cell's row; a planned cell that never recorded a result has no values, so no ratio uses its round.""" + recorded = path / "cell.json" + cell = json.loads(recorded.read_text()) if recorded.exists() else spec | {"exit_code": None, "missing": True} result_path = path / "greptimedb-1" / "load-result.json" result = json.loads(result_path.read_text()) if result_path.exists() else {} metrics_path = path / "metrics.prom" @@ -74,27 +76,24 @@ def load_cell(path: Path) -> dict: load_seconds = result.get("load_seconds") row = { "id": cell["id"], "round": cell["round"], "concurrency": cell["concurrency"], - "config": cell["config"], "exit_code": cell["exit_code"], "wal_device": cell["wal_device"], + "config": cell["config"], "exit_code": cell["exit_code"], "wal_device": cell.get("wal_device"), + "missing": cell.get("missing", False), "version": cell.get("version"), "load_wall_seconds": cell.get("load_wall_seconds"), "rows": result.get("count_after"), "load_seconds": load_seconds, "rows_per_second": result.get("rows_per_second") if cell["exit_code"] == 0 else None, "append_ack_p50_seconds": quantile(0.5, ack), "append_ack_p99_seconds": quantile(0.99, ack), } + # The counters register on their first event, so a scrape of an object store WAL + # that created objects omits the ones that never moved: those are zero, not unknown. + active = (cell.get("metrics_scraped") and cell.get("wal", {}).get("PROVIDER") == "experimental_object_store" + and metrics["samples"].get("created_objects_total")) for counter in COUNTERS: - row[counter] = metrics["samples"].get(counter) + row[counter] = metrics["samples"].get(counter, 0 if active else None) created = row["created_objects_total"] row["creates_per_minute"] = created * 60 / load_seconds if created is not None and load_seconds else None 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} @@ -124,10 +123,8 @@ def render(run: dict, rows: list, pairs: list) -> str: "| Stalled | Poisoned | Create failures | Conflicts | WAL on |", "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |"] for r in rows: - 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']})" + cell = f"{r['id']} (not run)" if r["missing"] else 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), @@ -149,12 +146,12 @@ def main() -> None: run = json.loads(run_path.read_text()) if run_path.exists() else {} plan_path = args.root / "plan.json" 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()} # 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}] + rows = [load_cell(args.root / "cells" / spec["id"], spec) for spec in planned] + rows += [load_cell(path) for path in sorted((args.root / "cells").glob("*")) + if path.name not in {spec["id"] for spec in planned} and (path / "cell.json").exists()] pairs = ratios(rows) - missing = [r["id"] for r in rows if r.get("missing")] + missing = [r["id"] for r in rows if r["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") diff --git a/.github/scripts/oswal-s3-bench.py b/.github/scripts/oswal-s3-bench.py index e648eb6b8dd..9df0d13dd27 100644 --- a/.github/scripts/oswal-s3-bench.py +++ b/.github/scripts/oswal-s3-bench.py @@ -18,11 +18,7 @@ # dependencies = ["boto3>=1.35,<2"] # /// -"""Plan, image tag, run manifest and bucket helpers of the object store WAL S3 benchmark. - -`plan`, `tag` and `manifest` need only the standard library; `probe` and -`cleanup` need boto3 and run under `uv run`. -""" +"""Helpers of the object store WAL S3 benchmark; only `probe` and `cleanup` need boto3 (`uv run`).""" from __future__ import annotations @@ -175,22 +171,20 @@ 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") + for name in ("timeout-minutes", "ttl-hours", "output"): + plan.add_argument(f"--{name}") commands.add_parser("tag").add_argument("requested") regions = commands.add_parser("regions") - regions.add_argument("--bucket", required=True) - regions.add_argument("--runner", required=True) + for name in ("bucket", "runner"): + regions.add_argument(f"--{name}", 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) - bucket.add_argument("--region", required=True) bucket.add_argument("--endpoint", default="") - bucket.add_argument("--prefix", required=True) + for option in ("bucket", "region", "prefix"): + bucket.add_argument(f"--{option}", required=True) args = parser.parse_args() if args.command == "plan": if args.timeout_minutes or args.ttl_hours: diff --git a/.github/workflows/oswal-s3-bench.yml b/.github/workflows/oswal-s3-bench.yml index 6a49580df23..8ad4d58e636 100644 --- a/.github/workflows/oswal-s3-bench.yml +++ b/.github/workflows/oswal-s3-bench.yml @@ -168,9 +168,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: | - uv run greptimedb/.github/scripts/oswal-s3-bench.py probe \ - --bucket "$S3_BUCKET" --region "$S3_REGION" --endpoint "$S3_ENDPOINT" --prefix "$RUN_PREFIX" + run: uv run greptimedb/.github/scripts/oswal-s3-bench.py probe --bucket "$S3_BUCKET" --region "$S3_REGION" --endpoint "$S3_ENDPOINT" --prefix "$RUN_PREFIX" - name: Put the Raft Engine WAL on the instance store run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh wal-disk "$WAL_ROOT" | tee "$RESULTS/wal-disk.txt" - name: Generate the dataset once @@ -188,9 +186,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: | - uv run greptimedb/.github/scripts/oswal-s3-bench.py cleanup \ - --bucket "$S3_BUCKET" --region "$S3_REGION" --endpoint "$S3_ENDPOINT" --prefix "$RUN_PREFIX" + run: uv run greptimedb/.github/scripts/oswal-s3-bench.py cleanup --bucket "$S3_BUCKET" --region "$S3_REGION" --endpoint "$S3_ENDPOINT" --prefix "$RUN_PREFIX" - name: Check the results hold no credentials if: always() env: diff --git a/tests/perf/test_oswal_s3_bench_scripts.py b/tests/perf/test_oswal_s3_bench_scripts.py index d9503ae24a8..26bcc443ef5 100644 --- a/tests/perf/test_oswal_s3_bench_scripts.py +++ b/tests/perf/test_oswal_s3_bench_scripts.py @@ -59,11 +59,10 @@ class PlanTest(unittest.TestCase): "2-1-E", "2-1-D", "2-1-R", "2-8-E", "2-8-D", "2-8-R"]) self.assertEqual(plan["generator_args"], ["--data-profile", "D"]) self.assertEqual((plan["rows"], plan["batch_rows"], plan["data_profile"]), (2_000_000, 1000, "D")) - by_config = {c["config"]: c["wal"] for c in plan["cells"]} - self.assertEqual(by_config["R"], {"PROVIDER": "raft_engine", "SYNC_WRITE": "true", "DIR": "/wal"}) - self.assertEqual(by_config["D"]["ACK_MODE"], "durable") - self.assertEqual(by_config["E"]["ACK_MODE"], "enqueued") - self.assertEqual(by_config["D"]["PROVIDER"], by_config["E"]["PROVIDER"]) + self.assertEqual({c["config"]: c["wal"] for c in plan["cells"]}, { + "R": {"PROVIDER": "raft_engine", "SYNC_WRITE": "true", "DIR": "/wal"}, + "D": {"PROVIDER": "experimental_object_store", "ACK_MODE": "durable"}, + "E": {"PROVIDER": "experimental_object_store", "ACK_MODE": "enqueued"}}) def test_smoke_overrides_rows_and_rounds(self): plan = bench.build_plan("8", "2000000", "1000", "2", smoke=True) @@ -115,13 +114,8 @@ class SummaryTest(unittest.TestCase): self.assertAlmostEqual(summary.quantile(0.99, buckets), 0.128 + 0.128 * 49 / 50) self.assertAlmostEqual(summary.quantile(0.25, buckets), 0.064 + 0.064 * 25 / 50) self.assertIsNone(summary.quantile(0.5, [])) - overflow = [(0.1, 1.0), (float("inf"), 4.0)] - self.assertEqual(summary.quantile(0.99, overflow), 0.1) - - def test_parse_keeps_only_object_store_wal_families(self): - parsed = summary.parse_metrics(ACK_BUCKETS) - self.assertEqual(parsed["samples"]["created_objects_total"], 120) - self.assertNotIn("kafka_client_produce_elapsed_count", parsed["samples"]) + self.assertEqual(summary.quantile(0.99, [(0.1, 1.0), (float("inf"), 4.0)]), 0.1) + self.assertNotIn("kafka_client_produce_elapsed_count", summary.parse_metrics(ACK_BUCKETS)["samples"]) def write_cell(self, root, cell_id, rows_per_second, exit_code=0, metrics=""): round_number, concurrency, config = cell_id.split("-") @@ -129,6 +123,7 @@ class SummaryTest(unittest.TestCase): (path / "greptimedb-1").mkdir(parents=True) (path / "cell.json").write_text(json.dumps({ "id": cell_id, "round": int(round_number), "concurrency": int(concurrency), "config": config, + "wal": bench.WAL_CONFIGS[config], "metrics_scraped": True, "exit_code": exit_code, "wal_device": "/dev/nvme1n1", "version": "1.4.0", "load_wall_seconds": 61.0})) if exit_code == 0: (path / "greptimedb-1" / "load-result.json").write_text(json.dumps({ @@ -154,7 +149,23 @@ class SummaryTest(unittest.TestCase): d_cell = next(c for c in result["cells"] if c["id"] == "1-8-D") self.assertEqual(d_cell["creates_per_minute"], 120.0) 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) + self.assertIn("| 1-8-D | 80 | 60.0 | 61.0 | 6,000 | 128.0 | 253.4 | 120 | 120.0 | 3 | 0 | 0 | 0 |", output) + r_cell = next(c for c in result["cells"] if c["id"] == "1-8-R") + self.assertIsNone(r_cell["stalled_appends_total"]) + + def test_untouched_counters_are_zero_only_for_an_active_object_store_wal(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + created = "greptime_logstore_object_store_wal_created_objects_total 40\n" + self.write_cell(root, "1-8-E", 90.0, metrics=created) + self.write_cell(root, "1-8-D", 80.0, metrics="") + cell_e, cell_d = (summary.load_cell(root / "cells" / i) for i in ("1-8-E", "1-8-D")) + self.assertEqual([cell_e[c] for c in summary.COUNTERS], [40, 0, 0, 0, 0]) + self.assertEqual([cell_d[c] for c in summary.COUNTERS], [None] * 5) + recorded = json.loads((root / "cells" / "1-8-E" / "cell.json").read_text()) + (root / "cells" / "1-8-E" / "cell.json").write_text(json.dumps(recorded | {"metrics_scraped": False})) + unscraped = summary.load_cell(root / "cells" / "1-8-E") + self.assertIsNone(unscraped["stalled_appends_total"]) def test_a_planned_round_that_never_ran_leaves_the_ratio_undefined(self): with tempfile.TemporaryDirectory() as directory: @@ -204,15 +215,17 @@ 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; } + [[ "$arg" != */metrics ]] || { + [[ "$FAKE_METRICS_RC" == 0 ]] || exit "$FAKE_METRICS_RC" + 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.""" +class CellLifecycleTest(unittest.TestCase): # the run script against fake docker and curl - def run_cells(self, root, load_rc): + def run_cells(self, root, load_rc, metrics_rc=0): bin_dir = root / "bin" bin_dir.mkdir() for name, body in (("docker", FAKE_DOCKER), ("curl", FAKE_CURL)): @@ -224,7 +237,7 @@ class CellLifecycleTest(unittest.TestCase): (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"), + "FAKE_LOAD_RC": str(load_rc), "FAKE_METRICS_RC": str(metrics_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", @@ -251,6 +264,19 @@ class CellLifecycleTest(unittest.TestCase): self.assertEqual(log.count("docker stop -t 120 fakecid"), 2) self.assertEqual(log.count("docker rm -fv fakecid"), 2) + def test_a_failed_scrape_fails_a_successful_load(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, log, _ = self.run_cells(root, load_rc=0, metrics_rc=22) + self.assertNotEqual(cells.returncode, 0) + for cell_id in ("1-8-D", "1-8-E"): + cell = root / "results" / "cells" / cell_id + recorded = json.loads((cell / "cell.json").read_text()) + self.assertEqual((recorded["exit_code"], recorded["metrics_scraped"]), (4, False)) + self.assertFalse((cell / "metrics.prom").exists()) + self.assertFalse((cell / "metrics.all").exists()) + 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) @@ -267,7 +293,8 @@ class CellLifecycleTest(unittest.TestCase): self.assertIn("GREPTIMEDB_STANDALONE__WAL__ACK_MODE=durable", config) self.assertIn("GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY=", config) recorded = json.loads((cell / "cell.json").read_text()) - self.assertEqual((recorded["exit_code"], recorded["version"]), (0, "1.4.0-test")) + self.assertEqual((recorded["exit_code"], recorded["version"], recorded["metrics_scraped"]), + (0, "1.4.0-test", True)) 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")