From c1a0ca9595c3a26a4273e85944927215383d7b7e Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Mon, 12 Oct 2026 13:36:55 +0000 Subject: [PATCH] ci: benchmark the object store WAL against S3 from an EC2 runner (#9536) * ci: benchmark the object store WAL against S3 from an EC2 runner Add a workflow_dispatch workflow that loads the o11ybench agent observability dataset into a fresh GreptimeDB container per cell and compares three WAL configurations that differ only in [wal]: Raft Engine with sync_write on the instance store, and the object store WAL in durable and enqueued mode, with the data files on S3 in the runner's region for all three. Rounds alternate the configuration order. A summary reports rows per second, D/R and E/R as the geometric mean over rounds, append acknowledgement p50 and p99, and the object store WAL counters. A smoke mode runs the same chain on ubuntu-latest against the MinIO fixture. Signed-off-by: jeremyhi * 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 * ci: drop redundant required flags from benchmark inputs Signed-off-by: jeremyhi * ci: fail a benchmark cell whose metrics scrape fails and zero untouched WAL counters A load that succeeds without its metrics scrape now fails the cell with exit code 4 and records metrics_scraped in cell.json. The summary reports an event counter that never registered as zero when the cell is an object store WAL cell whose scrape succeeded and created objects, and as unknown otherwise. Planned cells that never ran are read through the same path as recorded ones. Signed-off-by: jeremyhi * ci: register the S3 benchmark workflow for dispatch from its branch A workflow_dispatch workflow can only be dispatched once GitHub has registered it, which happens on the default branch or when one of its triggers fires. A push trigger limited to this branch registers it; validate runs only for workflow_dispatch, so a push run skips every job. Signed-off-by: jeremyhi * ci: delete the benchmark S3 prefix one object at a time boto3 now sends a CRC checksum instead of Content-MD5 with DeleteObjects, which the MinIO fixture of the smoke run rejects. DeleteObject needs neither and the prefix of one run holds at most a few thousand objects. Signed-off-by: jeremyhi * ci: default the S3 benchmark to c6id.2xlarge c7id is not offered in the runner region ap-southeast-1, where DescribeInstanceTypes rejects it. c6id is the compute optimized x86 family with a local NVMe instance store that the region offers, with the same 8 vCPU and 16 GiB at 2xlarge. Signed-off-by: jeremyhi * ci: drop the registration trigger of the S3 benchmark workflow The workflow is registered now, so dispatches no longer need a push trigger on its branch. Signed-off-by: jeremyhi * ci: fail an S3 benchmark run on EC2 without a local instance store Without an instance store the Raft Engine WAL of the R cells silently stayed on the root EBS volume, so the run measured a different baseline. Only a smoke run may keep it on the root volume now. Signed-off-by: jeremyhi * ci: let the instance store check fail its workflow step The step piped wal-disk into tee without pipefail, so a rejected host still passed. wal-disk now writes its record itself and the step runs it alone. Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- .github/scripts/oswal-s3-bench-run.sh | 179 ++++++++++++ .github/scripts/oswal-s3-bench-summary.py | 162 +++++++++++ .github/scripts/oswal-s3-bench.py | 222 +++++++++++++++ .github/workflows/oswal-s3-bench.yml | 229 ++++++++++++++++ tests/perf/test_oswal_s3_bench_scripts.py | 318 ++++++++++++++++++++++ 5 files changed, 1110 insertions(+) create mode 100644 .github/scripts/oswal-s3-bench-run.sh create mode 100644 .github/scripts/oswal-s3-bench-summary.py create mode 100644 .github/scripts/oswal-s3-bench.py create mode 100644 .github/workflows/oswal-s3-bench.yml create mode 100644 tests/perf/test_oswal_s3_bench_scripts.py diff --git a/.github/scripts/oswal-s3-bench-run.sh b/.github/scripts/oswal-s3-bench-run.sh new file mode 100644 index 00000000000..a9835c1e5f1 --- /dev/null +++ b/.github/scripts/oswal-s3-bench-run.sh @@ -0,0 +1,179 @@ +#!/usr/bin/env bash +# 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) + +# 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 "$@" +} + +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() { + : "${RESULTS:?}" + local root=$1 device + device=$(lsblk -dnpo NAME,MODEL | awk '/Instance Storage/ {print $1; exit}') + # Only a smoke run may keep the Raft Engine WAL on the root volume. + [[ -n "$device" || "${SMOKE:-}" == true ]] || { echo 'No local NVMe instance store; choose an instance type with one' >&2; return 1; } + sudo -n mkdir -p "$root" + if [[ -n "$device" ]]; then + sudo -n umount "$device" 2>/dev/null || true + sudo -n mkfs.ext4 -q -F "$device" + sudo -n mount -o noatime "$device" "$root" + fi + sudo -n chown "$(id -u):$(id -g)" "$root" + findmnt -no SOURCE,FSTYPE -T "$root" | tee "$RESULTS/wal-disk.txt" +} + +generate() { + : "${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_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 \ + > "$RESULTS/generator.log" 2>&1 + jq -e --argjson rows "$(jq .rows "$PLAN")" '.line_count == $rows' "$CORPUS_DIR/summary.json" >/dev/null +} + +# Scrape once, after the load succeeded or failed, while the server still runs. +scrape_metrics() { + 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() { + local rc=$? + trap - EXIT + set +e + if [[ -n "$cid" ]]; then + # 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" --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" +} + +# 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() { + : "${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= 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" + # 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") + 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" + done + # Pass the two keys by name: their values never reach argv, a file or the artifact. + export GREPTIMEDB_STANDALONE__STORAGE__ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID + export GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY=$AWS_SECRET_ACCESS_KEY + docker_env+=(-e GREPTIMEDB_STANDALONE__STORAGE__ACCESS_KEY_ID -e GREPTIMEDB_STANDALONE__STORAGE__SECRET_ACCESS_KEY) + printf 'GREPTIMEDB_STANDALONE__STORAGE__%s=\n' ACCESS_KEY_ID SECRET_ACCESS_KEY >> "$dir/config.env" + if jq -e '.wal.DIR' <<< "$spec" >/dev/null; then + wal_dir="$WAL_ROOT/$id"; mkdir -p "$wal_dir" + mounts=(--mount "type=bind,src=$wal_dir,dst=$(jq -r .wal.DIR <<< "$spec")") + wal_device=$(findmnt -no SOURCE -T "$wal_dir") + else + wal_device="s3://$S3_BUCKET/$RUN_PREFIX/$id/wal" + fi + 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[@]+"${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 + [[ $(docker inspect --format '{{.State.Running}}' "$cid") == true ]] || { echo "$id: server exited" >&2; exit 1; } + ((SECONDS < deadline)) || { echo "$id: server not healthy in 300s" >&2; exit 1; } + sleep 2 + done + version=$(docker exec "$cid" greptime --version | sed -n 's/^version: //p') + [[ -n "$version" ]] + started=$(date +%s.%N) + # Only the load phase, run exactly as run-agent-observability.sh runs it. + 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" \ + -e "TARGET_IMAGE=$GREPTIMEDB_IMAGE" -e "TARGET_VERSION=$version" -e "TARGET_DIGEST=${GREPTIMEDB_IMAGE##*@}" \ + -e "TARGET_ID=$name" -e TARGET_HOST=127.0.0.1 -e PREPARE_WORKERS= -e "MAX_IN_FLIGHT=$concurrency" \ + -e "GREPTIMEDB_BATCH_SIZE=$(jq -r .batch_rows "$PLAN")" --entrypoint /bin/bash \ + "$RUNTIME_IMAGE" scripts/agent-observability/greptimedb.sh load > "$dir/load.log" 2>&1 + ended=$(date +%s.%N) + 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 +} + +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) + : "${PLAN:?}" + failed=0 + 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 < <(jq -c '.cells[]' "$PLAN") + exit "$failed" ;; + scrub) scrub ;; + *) echo "usage: $0 check-runtime IMAGE | pull | wal-disk DIR | generate | cells | scrub" >&2; exit 2 ;; +esac diff --git a/.github/scripts/oswal-s3-bench-summary.py b/.github/scripts/oswal-s3-bench-summary.py new file mode 100644 index 00000000000..67186d422da --- /dev/null +++ b/.github/scripts/oswal-s3-bench-summary.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +# Copyright 2023 Greptime Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Summarize the object store WAL S3 benchmark as Markdown and summary.json.""" + +from __future__ import annotations + +import argparse +import json +import math +import re +from pathlib import Path + +PREFIX = "greptime_logstore_object_store_wal_" +COUNTERS = ("created_objects_total", "stalled_appends_total", "poisoned_total", + "create_failures_total", "create_conflicts_total") +SAMPLE = re.compile(r'([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+(\S+)') + + +def parse_metrics(text: str) -> dict: + """Map each unlabeled sample to its value and each histogram to its buckets.""" + samples, buckets = {}, {} + for line in text.splitlines(): + match = SAMPLE.fullmatch(line.strip()) + if not match or not match.group(1).startswith(PREFIX): + continue + name, labels, value = match.group(1)[len(PREFIX):], match.group(2) or "", float(match.group(3)) + le = re.search(r'le="([^"]+)"', labels) + if name.endswith("_bucket") and le: + buckets.setdefault(name[:-len("_bucket")], []).append((float(le.group(1)), value)) + else: + samples[name] = value + return {"samples": samples, "buckets": {name: sorted(items) for name, items in buckets.items()}} + + +def quantile(q: float, buckets: list) -> float | None: + """Prometheus histogram_quantile over cumulative (upper bound, count) buckets.""" + if not buckets or buckets[-1][1] == 0: + return None + rank = q * buckets[-1][1] + lower, below = 0.0, 0.0 + for upper, count in buckets: + if count >= rank: + if math.isinf(upper): + return lower + return lower + (upper - lower) * (rank - below) / (count - below) if count > below else upper + lower, below = upper, count + return lower + + +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, 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" + metrics = parse_metrics(metrics_path.read_text() if metrics_path.exists() else "") + ack = metrics["buckets"].get("append_ack_seconds", []) + 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.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, 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 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} + result = [] + for concurrency in sorted({r["concurrency"] for r in rows}): + rounds = sorted({r["round"] for r in rows if r["concurrency"] == concurrency}) + for config in ("D", "E"): + per_round = [] + for round_number in rounds: + base, value = speed.get((round_number, concurrency, "R")), speed.get((round_number, concurrency, config)) + per_round.append(value / base if base and value else None) + complete = [value for value in per_round if value is not None] + result.append({"concurrency": concurrency, "ratio": f"{config}/R", "per_round": per_round, + "geomean": geomean(complete) if len(complete) == len(per_round) else None}) + return result + + +def fmt(value, digits=0, scale=1.0) -> str: + return "—" if value is None else f"{value * scale:,.{digits}f}" + + +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')}` ({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 = 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), + *(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'] 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'])} " + f"| {fmt(p['geomean'], 3)} |") + return "\n".join(lines) + "\n" + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, required=True) + args = parser.parse_args() + args.root.mkdir(parents=True, exist_ok=True) + run_path = args.root / "run.json" + 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 [] + # The plan, not what happened to run, defines the expected cells and rounds. + 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["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") + + +if __name__ == "__main__": + main() diff --git a/.github/scripts/oswal-s3-bench.py b/.github/scripts/oswal-s3-bench.py new file mode 100644 index 00000000000..faad8ace674 --- /dev/null +++ b/.github/scripts/oswal-s3-bench.py @@ -0,0 +1,222 @@ +#!/usr/bin/env python3 +# Copyright 2023 Greptime Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# /// script +# requires-python = ">=3.10" +# dependencies = ["boto3>=1.35,<2"] +# /// + +"""Helpers of the object store WAL S3 benchmark; only `probe` and `cleanup` need boto3 (`uv run`).""" + +from __future__ import annotations + +import argparse +import json +import os +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. +DATA_PROFILES = {5_000: "S", 2_000_000: "D", 10_000_000: "P", 100_000_000: "M"} +# Identical [storage] for every configuration; only [wal] differs. +WAL_CONFIGS = { + "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"}, +} +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_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") + + +def positive(name: str, value: str, upper: int) -> int: + if not re.fullmatch(r"[1-9][0-9]*", value) or int(value) > upper: + raise ValueError(f"{name} must be an integer in 1..{upper}, got {value!r}") + 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): + raise ValueError(f"concurrency repeats a level: {concurrency!r}") + row_count = SMOKE_ROWS if smoke else positive("rows", rows, 100_000_000) + round_count = 1 if smoke else positive("rounds", rounds, 10) + profile = DATA_PROFILES.get(row_count) + cells = [] + for round_number in range(1, round_count + 1): + # Alternate the order between rounds so no configuration always runs first. + for level in levels: + for config in ORDERS[(round_number - 1) % 2]: + cells.append({ + "id": f"{round_number}-{level}-{config}", "round": round_number, + "concurrency": level, "config": config, "wal": WAL_CONFIGS[config], + }) + return { + "smoke": smoke, "rows": row_count, "rounds": round_count, + "batch_rows": positive("batch_rows", batch_rows, 100_000), "concurrency": levels, + "data_profile": profile or "custom", + "generator_args": ["--data-profile", profile] if profile else ["--line-count", str(row_count)], + "cells": cells, + } + + +def newest_nightly(tags: list[str]) -> str: + dated = [(match.group(1), tag) for tag in tags if (match := NIGHTLY.fullmatch(tag))] + if not dated: + raise ValueError("no v*-nightly-* tag found") + return max(dated)[1] + + +def resolve_tag(requested: str) -> str: + if requested != "nightly": + if not TAG.fullmatch(requested): + raise ValueError(f"invalid image tag: {requested!r}") + return requested + url = ("https://hub.docker.com/v2/repositories/greptime/greptimedb/tags" + "?page_size=100&name=nightly&ordering=last_updated") + with urllib.request.urlopen(url, timeout=30) as response: + return newest_nightly([tag["name"] for tag in json.load(response)["results"]]) + + +def s3_client(args): + import boto3 + from botocore.config import Config + + return boto3.client("s3", region_name=args.region, endpoint_url=args.endpoint or None, + config=Config(retries={"mode": "standard", "max_attempts": 5})) + + +def probe(args) -> None: + client, key = s3_client(args), f"{args.prefix}/probe/object" + deadline = time.monotonic() + 60 + while True: + try: + client.put_object(Bucket=args.bucket, Key=key, Body=b"probe") + break + except Exception: + # A fresh MinIO creates its bucket after the container starts. + if time.monotonic() > deadline: + raise + time.sleep(3) + listed = client.list_objects_v2(Bucket=args.bucket, Prefix=f"{args.prefix}/probe/") + if [item["Key"] for item in listed.get("Contents", [])] != [key]: + raise RuntimeError(f"listing {args.prefix}/probe/ did not return exactly {key}") + client.delete_object(Bucket=args.bucket, Key=key) + print(f"put, list and delete succeeded under s3://{args.bucket}/{args.prefix}/probe/") + + +def cleanup(args) -> None: + # One DeleteObject per key: DeleteObjects needs a checksum header the MinIO fixture rejects. + client, deleted = s3_client(args), 0 + for page in client.get_paginator("list_objects_v2").paginate(Bucket=args.bucket, Prefix=f"{args.prefix}/"): + for item in page.get("Contents", []): + client.delete_object(Bucket=args.bucket, Key=item["Key"]) + deleted += 1 + print(f"deleted {deleted} objects under s3://{args.bucket}/{args.prefix}/") + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + plan = commands.add_parser("plan") + for name in ("concurrency", "rows", "batch-rows", "rounds"): + plan.add_argument(f"--{name}", required=True) + plan.add_argument("--smoke", choices=("true", "false"), required=True) + 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") + 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("--endpoint", default="") + 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: + 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: + json.dump(result, output, indent=2) + print(f"{len(result['cells'])} cells, {result['rows']} rows, profile {result['data_profile']}") + elif args.command == "tag": + tag = resolve_tag(args.requested) + 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/: {args.prefix!r}") + (probe if args.command == "probe" else cleanup)(args) + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except ValueError as error: + print(f"::error::{error}", file=sys.stderr) + sys.exit(1) diff --git a/.github/workflows/oswal-s3-bench.yml b/.github/workflows/oswal-s3-bench.yml new file mode 100644 index 00000000000..c31dfe8f1c3 --- /dev/null +++ b/.github/workflows/oswal-s3-bench.yml @@ -0,0 +1,229 @@ +name: Object Store WAL S3 Benchmark +on: + workflow_dispatch: + inputs: + smoke: + description: 'Smoke run: ubuntu-latest and the MinIO fixture instead of EC2 and S3, 50000 rows, one round; its numbers mean nothing' + type: boolean + default: false + greptimedb_tag: + description: 'Docker Hub greptime/greptimedb tag; nightly = the newest v*-nightly-* tag (latest is a release without the object store WAL)' + type: string + default: nightly + instance_type: + description: EC2 instance type in the bucket region; the d of c6id gives a local NVMe instance store for the Raft Engine WAL + type: string + default: c6id.2xlarge + concurrency: + description: Comma-separated loader concurrency levels + type: string + default: '1,8' + rows: + description: Dataset rows (2000000 = o11ybench profile D) + type: number + default: 2000000 + batch_rows: + description: Rows per write request + type: number + default: 1000 + rounds: + description: Rounds; the configuration order alternates between rounds (R, D, E then E, D, R) + type: number + default: 2 + runtime_image: + description: 'o11ybench runtime image: auto = AWS ECR Public; explicit tag/digest overrides' + type: string + default: auto + benchmark_timeout_minutes: + description: Benchmark job timeout in minutes (1..360) + type: number + default: 240 + janitor_ttl_hours: + description: Instance lifetime before janitor eligibility (1..168 hours; at least timeout + 2h) + type: number + default: 8 +permissions: + contents: read +env: + SMOKE: ${{ inputs.smoke }} + GREPTIMEDB_TAG: ${{ inputs.greptimedb_tag }} + REQUESTED_INSTANCE_TYPE: ${{ inputs.instance_type }} + CONCURRENCY: ${{ inputs.concurrency }} + ROWS: ${{ inputs.rows }} + BATCH_ROWS: ${{ inputs.batch_rows }} + ROUNDS: ${{ inputs.rounds }} + RUNTIME_IMAGE: ${{ inputs.runtime_image != 'auto' && inputs.runtime_image || 'public.ecr.aws/i8k6a5e1/greptime/o11ybenchruntime:77a84e1c8136@sha256:c482993ec177b026235d6cd0c61d4fb6d6f1088bc78bdc0e5964a5398de6dba0' }} + BENCHMARK_TIMEOUT_MINUTES: ${{ inputs.benchmark_timeout_minutes }} + JANITOR_TTL_HOURS: ${{ inputs.janitor_ttl_hours }} +jobs: + validate: + runs-on: ubuntu-latest + timeout-minutes: 15 + outputs: + instance_type: ${{ steps.runner.outputs.instance_type }} + greptimedb_tag: ${{ steps.plan.outputs.greptimedb_tag }} + runtime_image: ${{ steps.runtime.outputs.runtime_image }} + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - name: Test benchmark scripts + run: | + python3 greptimedb/tests/perf/test_oswal_s3_bench_scripts.py + python3 greptimedb/tests/perf/test_ci_runner_scripts.py + - 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" \ + --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 + run: python3 greptimedb/.github/scripts/ci-runner-config.py --provider AWS --instance-type "$REQUESTED_INSTANCE_TYPE" + - name: Check bucket and runner regions + env: + BUCKET_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + RUNNER_REGION: ${{ vars.EC2_RUNNER_REGION }} + run: python3 greptimedb/.github/scripts/oswal-s3-bench.py regions --bucket "$BUCKET_REGION" --runner "$RUNNER_REGION" + - name: Check the runtime image + id: runtime + run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh check-runtime "$RUNTIME_IMAGE" + provision: + needs: validate + if: ${{ !inputs.smoke }} + runs-on: ubuntu-latest + timeout-minutes: 45 + outputs: + label: ${{ steps.aws.outputs.label }} + instance_id: ${{ steps.aws.outputs.instance_id }} + runner_name: ${{ steps.aws.outputs.runner_name }} + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - uses: aws-actions/configure-aws-credentials@v4 + with: + aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + aws-region: ${{ vars.EC2_RUNNER_REGION }} + - uses: astral-sh/setup-uv@v6 + - name: Create ephemeral EC2 runner + id: aws + env: + GH_PERSONAL_ACCESS_TOKEN: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }} + AWS_REGION: ${{ vars.EC2_RUNNER_REGION }} + AWS_EC2_IMAGE_ID: ${{ vars.BENCHMARK_EC2_IMAGE_ID || 'auto' }} + AWS_EC2_INSTANCE_TYPE: ${{ needs.validate.outputs.instance_type }} + AWS_EC2_SUBNET_ID: ${{ vars.EC2_RUNNER_SUBNET_ID }} + AWS_EC2_SECURITY_GROUP_ID: ${{ vars.EC2_RUNNER_SECURITY_GROUP_ID }} + AWS_EC2_SYSTEM_DISK_GIB: '60' + AWS_EC2_TTL_HOURS: ${{ inputs.janitor_ttl_hours }} + AWS_EC2_RUN_ID: ${{ github.run_id }}-${{ github.run_attempt }} + run: uv run greptimedb/.github/scripts/aws-ec2-runner-provision.py + benchmark: + needs: + - validate + - provision + if: ${{ !cancelled() && needs.validate.result == 'success' && (inputs.smoke || needs.provision.result == 'success') }} + runs-on: ${{ inputs.smoke && 'ubuntu-latest' || needs.provision.outputs.label }} + timeout-minutes: ${{ fromJSON(format('{0}', inputs.benchmark_timeout_minutes)) }} + env: + INSTANCE_TYPE: ${{ inputs.smoke && 'ubuntu-latest' || needs.validate.outputs.instance_type }} + GREPTIMEDB_TAG: ${{ needs.validate.outputs.greptimedb_tag }} + RUNTIME_IMAGE: ${{ needs.validate.outputs.runtime_image }} + RUN_PREFIX: oswal-bench/${{ github.run_id }}-${{ github.run_attempt }} + OWNER: o11ybench-oswal-${{ github.run_id }}-${{ github.run_attempt }} + RESULTS: ${{ github.workspace }}/oswal-results + CORPUS_DIR: ${{ github.workspace }}/oswal-corpus + PLAN: ${{ github.workspace }}/oswal-results/plan.json + WAL_ROOT: /mnt/oswal-wal + S3_BUCKET: ${{ inputs.smoke && 'greptime' || vars.AWS_CI_TEST_BUCKET }} + S3_REGION: ${{ inputs.smoke && 'us-west-2' || vars.AWS_CI_TEST_BUCKET_REGION }} + S3_ENDPOINT: ${{ inputs.smoke && 'http://127.0.0.1:9000' || '' }} + S3_CONTAINER_ENDPOINT: ${{ inputs.smoke && 'http://host.docker.internal:9000' || '' }} + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - name: Set up benchmark dependencies + uses: ./greptimedb/.github/actions/setup-benchmark + - uses: astral-sh/setup-uv@v6 + - name: Start the MinIO fixture + if: inputs.smoke + run: | + bash greptimedb/.github/scripts/pull-benchmark-image.sh greptime/minio:2024 + docker compose -f greptimedb/tests-integration/fixtures/docker-compose.yml up -d --wait minio + - name: Pull images and write the plan + run: | + 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" + 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 }} + 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" + - name: Put the Raft Engine WAL on the instance store + run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh wal-disk "$WAL_ROOT" + - name: Generate the dataset once + run: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh generate + - name: Load every cell + 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: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh cells + - name: Summary + if: always() + run: python3 greptimedb/.github/scripts/oswal-s3-bench-summary.py --root "$RESULTS" >> "$GITHUB_STEP_SUMMARY" + - name: Delete the run's S3 prefix + if: always() + 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" + - name: Check the results hold no credentials + if: always() + 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: bash greptimedb/.github/scripts/oswal-s3-bench-run.sh scrub + - name: Upload results (no corpus) + if: always() + uses: actions/upload-artifact@v4 + with: + name: oswal-s3-bench-${{ inputs.smoke && 'smoke' || 'ec2' }}-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ env.RESULTS }} + if-no-files-found: warn + retention-days: 14 + teardown: + needs: + - provision + - benchmark + if: ${{ always() && !inputs.smoke && (needs.provision.outputs.instance_id != '' || needs.provision.result != 'skipped') }} + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - uses: aws-actions/configure-aws-credentials@v4 + with: + aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + aws-region: ${{ vars.EC2_RUNNER_REGION }} + - uses: astral-sh/setup-uv@v6 + - name: Delete EC2 and unregister runner + env: + GH_PERSONAL_ACCESS_TOKEN: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }} + AWS_REGION: ${{ vars.EC2_RUNNER_REGION }} + AWS_EC2_INSTANCE_ID: ${{ needs.provision.outputs.instance_id }} + AWS_EC2_RUNNER_NAME: ${{ needs.provision.outputs.runner_name }} + AWS_EC2_RUN_ID: ${{ github.run_id }}-${{ github.run_attempt }} + run: uv run greptimedb/.github/scripts/aws-ec2-runner-teardown.py diff --git a/tests/perf/test_oswal_s3_bench_scripts.py b/tests/perf/test_oswal_s3_bench_scripts.py new file mode 100644 index 00000000000..dc81240f1ae --- /dev/null +++ b/tests/perf/test_oswal_s3_bench_scripts.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python3 +# Copyright 2023 Greptime Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Offline tests of the object store WAL S3 benchmark plan and summary.""" + +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +SCRIPTS = Path(__file__).parents[2] / ".github/scripts" + + +def load(name, filename): + spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +bench = load("oswal_s3_bench", "oswal-s3-bench.py") +summary = load("oswal_s3_bench_summary", "oswal-s3-bench-summary.py") + +ACK_BUCKETS = """# TYPE greptime_logstore_object_store_wal_append_ack_seconds histogram +greptime_logstore_object_store_wal_append_ack_seconds_bucket{le="0.064"} 0 +greptime_logstore_object_store_wal_append_ack_seconds_bucket{le="0.128"} 50 +greptime_logstore_object_store_wal_append_ack_seconds_bucket{le="0.256"} 100 +greptime_logstore_object_store_wal_append_ack_seconds_bucket{le="+Inf"} 100 +greptime_logstore_object_store_wal_append_ack_seconds_sum 12 +greptime_logstore_object_store_wal_append_ack_seconds_count 100 +greptime_logstore_object_store_wal_created_objects_total 120 +greptime_logstore_object_store_wal_stalled_appends_total 3 +greptime_logstore_object_store_wal_poisoned_total 0 +greptime_logstore_kafka_client_produce_elapsed_count 9 +""" + + +class PlanTest(unittest.TestCase): + def test_default_plan_alternates_the_order_between_rounds(self): + plan = bench.build_plan("1,8", "2000000", "1000", "2", smoke=False) + self.assertEqual([c["id"] for c in plan["cells"]], [ + "1-1-R", "1-1-D", "1-1-E", "1-8-R", "1-8-D", "1-8-E", + "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")) + 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) + self.assertEqual((plan["rows"], plan["rounds"], plan["data_profile"]), (50_000, 1, "custom")) + self.assertEqual(plan["generator_args"], ["--line-count", "50000"]) + self.assertEqual([c["id"] for c in plan["cells"]], ["1-8-R", "1-8-D", "1-8-E"]) + + def test_invalid_inputs_are_rejected(self): + for args in (("0", "10", "10", "1"), ("1,1", "10", "10", "1"), ("1", "abc", "10", "1"), + ("1", "10", "10", "11"), ("1,", "10", "10", "1")): + with self.subTest(args=args), self.assertRaises(ValueError): + bench.build_plan(*args, smoke=False) + + def test_nightly_tag_is_the_newest_by_date(self): + tags = ["latest", "v1.3.0-alpha.1-nightly-20260907", "v1.4.0-nightly-20261012", + "v1.4.0-nightly-20261005", "nightly-arm64"] + self.assertEqual(bench.newest_nightly(tags), "v1.4.0-nightly-20261012") + with self.assertRaises(ValueError): + bench.newest_nightly(["latest"]) + self.assertEqual(bench.resolve_tag("v1.4.0-nightly-20261012"), "v1.4.0-nightly-20261012") + with self.assertRaises(ValueError): + bench.resolve_tag("bad tag") + + 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, "c6id.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): + def test_quantile_interpolates_like_prometheus(self): + buckets = summary.parse_metrics(ACK_BUCKETS)["buckets"]["append_ack_seconds"] + self.assertAlmostEqual(summary.quantile(0.5, buckets), 0.128) + 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, [])) + 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("-") + path = root / "cells" / cell_id + (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({ + "count_after": 6000, "load_seconds": 60.0, "rows_per_second": rows_per_second})) + (path / "metrics.prom").write_text(metrics) + + def test_summary_reports_cells_and_geometric_mean_ratios(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)) + (root / "run.json").write_text(json.dumps({"rows": 6000, "batch_rows": 1000, "instance_type": "c6id.2xlarge"})) + speeds = {"1-8-R": 100.0, "1-8-D": 80.0, "1-8-E": 90.0, "2-8-E": 99.0, "2-8-D": 50.0, "2-8-R": 110.0} + for cell_id, speed in speeds.items(): + self.write_cell(root, cell_id, speed, metrics=ACK_BUCKETS if cell_id.endswith("D") else "") + 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([c["id"] for c in result["cells"]], [c["id"] for c in plan["cells"]]) + d_ratio = next(r for r in result["ratios"] if r["ratio"] == "D/R") + self.assertAlmostEqual(d_ratio["geomean"], (0.8 * (50 / 110)) ** 0.5) + self.assertAlmostEqual(next(r for r in result["ratios"] if r["ratio"] == "E/R")["geomean"], 0.9) + 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 | 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: + 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) + self.write_cell(root, "1-1-R", 100.0) + self.write_cell(root, "1-1-D", 0.0, exit_code=1) + rows = [summary.load_cell(path) for path in sorted((root / "cells").iterdir())] + ratio = next(r for r in summary.ratios(rows) if r["ratio"] == "D/R") + self.assertEqual((ratio["per_round"], ratio["geomean"]), ([None], None)) + 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 ]] || { + [[ "$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): # the run script against fake docker and curl + + 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)): + (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_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", + "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_an_ec2_run_without_an_instance_store_fails(self): + with tempfile.TemporaryDirectory() as directory: + (Path(directory) / "lsblk").write_text("#!/usr/bin/env bash\necho '/dev/nvme0n1 Amazon Elastic Block Store'\n") + (Path(directory) / "lsblk").chmod(0o755) + env = os.environ | {"PATH": f"{directory}:{os.environ['PATH']}", "SMOKE": "false", "RESULTS": directory} + done = subprocess.run(["bash", str(SCRIPTS / "oswal-s3-bench-run.sh"), "wal-disk", f"{directory}/wal"], + env=env, capture_output=True, text=True, timeout=60) + self.assertNotEqual(done.returncode, 0) + self.assertIn("No local NVMe instance store", done.stderr) + + 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) + 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=", config) + recorded = json.loads((cell / "cell.json").read_text()) + 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") + 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()