mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-19 10:22:31 +00:00
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 <fengjiachun@gmail.com> * 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> * ci: drop redundant required flags from benchmark inputs Signed-off-by: jeremyhi <fengjiachun@gmail.com> * 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 <fengjiachun@gmail.com> * 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 <fengjiachun@gmail.com> * 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 <fengjiachun@gmail.com> * 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 <fengjiachun@gmail.com> * 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 <fengjiachun@gmail.com> * 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 <fengjiachun@gmail.com> * 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 <fengjiachun@gmail.com> --------- Signed-off-by: jeremyhi <fengjiachun@gmail.com>
This commit is contained in:
jeremyhi
@@ -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=<redacted>\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
|
||||
@@ -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()
|
||||
@@ -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/<run>: {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)
|
||||
Reference in New Issue
Block a user