From 0e82b29b23c0c4349954b11c7fe9df8de86650e9 Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Fri, 16 Oct 2026 09:57:50 +0000 Subject: [PATCH] ci: add crash, restart, multi-region and drop cells to the S3 WAL benchmark (#9632) * ci: add crash, restart, multi-region and drop cells to the S3 WAL benchmark Extend the object store WAL S3 benchmark beyond one load per configuration: - Inputs for a full image reference, the configurations to run (R0 adds Raft Engine with the standalone default sync_write = false), extra environment for every server container, a restart after each load, and a JSON list of named cell groups that override those inputs. - Restart cells SIGKILL the loaded server, restart it on the same prefix and data home, and time /health, the first correct count and the log store recovery; min_objects fails a cell that left too few objects indexed. - Crash groups write striped 1000-row batches from several writers, kill the server at a random point, restart it and check every acknowledged batch against the table (durable: present exactly once; enqueued: no duplicate, no partial batch, losses counted). A pinned variant keeps one row of a second table unflushed and samples the indexed objects while it runs. - Idle windows scrape the server before and after a quiet period. - A load group may split the loaded table into equal event_time ranges, so every batch writes to every region, and checks that each region took rows. - Drop groups write a table, drop it, kill and restart the server, then write and flush another table and wait for the dropped table's objects to be collected. - The summary adds WAL traffic, objects per second, mean object size, entries and appends per object, S3 requests per operation, server and host CPU, the loader's own request latency, rows per region, crash cycles, and each restart's replay per region from the server log. Signed-off-by: jeremyhi * ci: report the object sequences a replay spans instead of objects replayed The width of a region's replayed entry range counts every object sequence in it, including objects that hold no entry of the region and are skipped, so the summary labels it as a span. Signed-off-by: jeremyhi * ci: fail a crash cycle that loses a batch an earlier restart recovered The oracle accepted every missing acknowledged batch in the enqueued mode, so a batch that one restart recovered and a later restart no longer found passed. Every cycle keeps what its restart found, and a cycle now fails in either mode unless every batch an earlier restart recovered is still present, so only the current cycle's backlog that was not durable at its kill may be missing. Signed-off-by: jeremyhi * ci: keep the pinned WAL pinned and label OpenDAL operations as such The pin table inherited the engine's 10-minute auto flush, so a longer pinned cycle could flush its row and release the WAL before the kill. It now sets its own auto_flush_interval of one day, longer than any crash group can run, and a pinned cycle fails unless its indexed objects never fall while it runs. The summary's request counts come from opendal_operation_duration_seconds, which counts OpenDAL calls rather than HTTP requests (a paginated list or a multipart write is one operation), so the fields and columns now say S3 operations. Signed-off-by: jeremyhi * ci: check a pinned cycle's retention by the collected-object counter Net growth of the indexed objects can hide collection, and short cycles take no periodic sample, so a pinned cycle now fails unless deleted_objects_total is the same in the scrape before its writers start and in the one before the kill, both of the same server process. The operation columns name OpenDAL. Signed-off-by: jeremyhi * ci: let a pinned cell collect only the objects below its pin The first write after a start lets the store collect the empty start object asynchronously, so a deletion after a cycle's baseline does not mean the pin let go. A pinned cell now scrapes the server before it writes the pin row and fails only once its servers together have collected more objects than were indexed then, counting the first server from that scrape and every restarted one from its start. Signed-off-by: jeremyhi * ci: test that a pinned cell's allowance spans its server restarts Two servers that each collect fewer objects than lie below the pin but more together now fail the cell in the second cycle. Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- .github/scripts/oswal-s3-bench-crash.py | 229 +++++++ .github/scripts/oswal-s3-bench-load.sh | 27 + .github/scripts/oswal-s3-bench-run.sh | 341 ++++++++-- .github/scripts/oswal-s3-bench-summary.py | 368 +++++++++- .github/scripts/oswal-s3-bench.py | 236 ++++++- .github/workflows/oswal-s3-bench.yml | 35 +- tests/perf/test_oswal_s3_bench_scripts.py | 790 +++++++++++++++++++++- 7 files changed, 1907 insertions(+), 119 deletions(-) create mode 100644 .github/scripts/oswal-s3-bench-crash.py create mode 100644 .github/scripts/oswal-s3-bench-load.sh diff --git a/.github/scripts/oswal-s3-bench-crash.py b/.github/scripts/oswal-s3-bench-crash.py new file mode 100644 index 00000000000..35b382aa24b --- /dev/null +++ b/.github/scripts/oswal-s3-bench-crash.py @@ -0,0 +1,229 @@ +#!/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. + +"""Writer and oracle of the benchmark's crash cells. + +Every writer owns the stripe `first + index, first + index + writers, ...` of +batch sequence numbers, inserts `rows` rows per batch and stops at the first +statement that is not acknowledged, so a stripe has at most one attempted but +unacknowledged batch: the one in flight at the kill. Every request is logged +with its status and the time it ended, on the same clock as the kill stamps. +With `--seconds` the writers also stop once that long has passed, and the +command fails unless every writer wrote until then without a failure. +""" + +from __future__ import annotations + +import argparse +import collections +import json +import os +import sys +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + + +def insert(url: str, table: str, seq: int, rows: int) -> str: + values = ", ".join(f"({seq}, {seq * rows + i})" for i in range(rows)) + body = urllib.parse.urlencode({"sql": f"INSERT INTO {table} (seq, ts) VALUES {values}"}).encode() + try: + with urllib.request.urlopen(urllib.request.Request(url, data=body), timeout=60) as response: + return str(response.status) + except urllib.error.HTTPError as error: + return str(error.code) + except (OSError, urllib.error.URLError): + # The connection dropped: the request may or may not have been applied. + return "000" + + +def writer(args, index: int, deadline: float, failures: list) -> None: + directory, seq = Path(args.dir), args.first_seq + index + with open(directory / f"attempted-{index}.log", "a") as attempted, \ + open(directory / f"requests-{index}.log", "a") as requests, \ + open(directory / f"acked-{index}.log", "a") as acked: + while time.monotonic() < deadline: + attempted.write(f"{seq}\n") + attempted.flush() + code = insert(args.url, args.table, seq, args.rows) + requests.write(f"{seq} {code} {time.time_ns()}\n") + requests.flush() + if code != "200": + failures.append(f"writer {index}: batch {seq} ended with {code}") + break + acked.write(f"{seq}\n") + acked.flush() + seq += args.writers + (directory / f"exit-{index}.log").write_text(f"{time.time_ns()}\n") + + +def write(args) -> int: + os.makedirs(args.dir, exist_ok=True) + deadline = time.monotonic() + args.seconds if args.seconds else float("inf") + failures = [] + + def run(index): + try: + writer(args, index, deadline, failures) + except Exception as error: + failures.append(f"writer {index}: {error!r}") + raise + + threads = [threading.Thread(target=run, args=(i,)) for i in range(args.writers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + # Crash cells end their writers with a kill; timed writers must have run to the deadline. + if args.seconds and (failures or not any(lines(Path(args.dir) / f"acked-{i}.log") for i in range(args.writers))): + print("; ".join(failures) or "no batch was acknowledged", file=sys.stderr) + return 1 + return 0 + + +def lines(path: Path) -> list: + return [line.split() for line in path.read_text().splitlines() if line.strip()] if path.exists() else [] + + +def found_batches(path: Path) -> dict: + """GreptimeDB's JSON answer to SELECT seq, count(*) ... GROUP BY seq, as batch -> row count.""" + return {int(seq): int(count) for seq, count in json.loads(path.read_text())["output"][0]["records"]["rows"]} + + +def check_kill(directory: Path, writers: int, before_ns: int, after_ns: int, window_max_ms: int) -> dict: + """A dropped connection is in flight when it ended at or after the stamp taken before the kill.""" + errors, in_flight, running = [], 0, 0 + if after_ns - before_ns > window_max_ms * 1_000_000: + errors.append(f"kill window {after_ns - before_ns}ns exceeds {window_max_ms}ms") + for index in range(writers): + exit_path = directory / f"exit-{index}.log" + if not exit_path.exists(): + errors.append(f"writer {index} left no exit time") + continue + running += int(exit_path.read_text()) >= before_ns + for seq, code, end_ns in lines(directory / f"requests-{index}.log"): + if code == "200": + continue + if code != "000": + errors.append(f"writer {index}: batch {seq} rejected with HTTP {code}") + elif int(end_ns) < before_ns: + errors.append(f"writer {index}: batch {seq} lost its connection before the kill") + else: + in_flight += 1 + if running == 0: + errors.append("no writer was running at the kill") + return {"kill_window_ns": after_ns - before_ns, "in_flight": in_flight, "writers_at_kill": running, + "kill_errors": errors} + + +def verify(run_dir: Path, found: dict, rows: int, enqueued: bool, cycle: int) -> dict: + """Checks the table, read as batch -> row count, against every cycle's writers so far and + against what every earlier restart recovered, and counts what the given cycle's writers + acknowledged, left and lost on their own.""" + acked, stripes, cycle_attempted, cycle_acked = [], collections.OrderedDict(), set(), set() + recovered_before = set() + for cycle_dir in sorted(run_dir.glob("cycle-*"), key=lambda p: int(p.name.split("-")[1])): + current = cycle_dir.name == f"cycle-{cycle}" + if int(cycle_dir.name.split("-")[1]) < cycle and (cycle_dir / "found.json").exists(): + recovered_before.update(found_batches(cycle_dir / "found.json")) + for path in sorted(cycle_dir.glob("attempted-*.log")): + stripes[(cycle_dir.name, path.name)] = [int(seq) for (seq,) in lines(path)] + if current: + cycle_attempted.update(stripes[(cycle_dir.name, path.name)]) + for path in sorted(cycle_dir.glob("acked-*.log")): + acked += [int(seq) for (seq,) in lines(path)] + if current: + cycle_acked.update(int(seq) for (seq,) in lines(path)) + acked_set = set(acked) + attempted = {seq for stripe in stripes.values() for seq in stripe} + errors = [] + duplicated = sorted(seq for seq, count in found.items() if count > rows) + partial = sorted(seq for seq, count in found.items() if count < rows) + if duplicated: + errors.append(f"batches present more than once: {duplicated[:20]}") + if partial: + errors.append(f"batches present in part: {partial[:20]}") + missing = sorted(acked_set - set(found)) + if missing and not enqueued: + errors.append(f"acknowledged but absent: {missing[:20]}") + # A batch an earlier restart recovered was durable then, so in either mode only the + # current cycle's backlog that was not durable at its kill may be missing. + vanished = sorted(recovered_before - set(found)) + if vanished: + errors.append(f"recovered by an earlier restart but absent: {vanished[:20]}") + unattempted = sorted(set(found) - attempted) + if unattempted: + errors.append(f"present but never attempted: {unattempted[:20]}") + # The unacknowledged tail of each stripe was in flight at a kill. + in_flight = set() + for key, stripe in stripes.items(): + prefix = 0 + while prefix < len(stripe) and stripe[prefix] in acked_set: + prefix += 1 + if any(seq in acked_set for seq in stripe[prefix:]): + errors.append(f"{key}: acknowledged after a failure") + in_flight.update(stripe[prefix:]) + extra = sorted(set(found) - acked_set) + stray = [seq for seq in extra if seq not in in_flight] + if stray and not enqueued: + errors.append(f"present, unacknowledged and not in flight at a kill: {stray[:20]}") + if len(acked) != len(acked_set): + errors.append("a batch was acknowledged twice") + present = set(found) & cycle_attempted + return {"cycle_acked_batches": len(cycle_acked), "cycle_present_batches": len(present), + "cycle_lost_batches": len(cycle_acked - present), "cycle_lost_rows": len(cycle_acked - present) * rows, + "cycle_extra_batches": len(present - cycle_acked), + "total_found_batches": len(found), "total_acked_batches": len(acked_set), + "total_lost_batches": len(missing), "total_extra_batches": len(extra), "verify_errors": errors} + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + w = commands.add_parser("write") + w.add_argument("--dir", required=True) + w.add_argument("--url", required=True) + w.add_argument("--table", default="crash") + for name in ("writers", "first-seq", "rows"): + w.add_argument(f"--{name}", type=int, required=True) + w.add_argument("--seconds", type=float, default=0, help="stop after this long; 0 writes until a failure") + c = commands.add_parser("check") + c.add_argument("--run-dir", type=Path, required=True) + c.add_argument("--cycle", type=int, required=True) + c.add_argument("--found", type=Path, required=True, help="GreptimeDB JSON output of SELECT seq, count(*)") + c.add_argument("--mode", choices=("durable", "enqueued"), required=True) + for name in ("writers", "rows", "before-ns", "after-ns", "window-max-ms"): + c.add_argument(f"--{name}", type=int, required=True) + c.add_argument("--pin-rows", type=int, help="rows of the pin table, whose one row was acknowledged first") + args = parser.parse_args() + if args.command == "write": + return write(args) + result = {"cycle": args.cycle} + result |= check_kill(args.run_dir / f"cycle-{args.cycle}", args.writers, args.before_ns, args.after_ns, + args.window_max_ms) + result |= verify(args.run_dir, found_batches(args.found), args.rows, + args.mode == "enqueued", args.cycle) + if args.pin_rows is not None and (args.pin_rows > 1 or args.mode == "durable" and args.pin_rows != 1): + result["verify_errors"].append(f"the pin table holds {args.pin_rows} rows, not 1") + result["passed"] = not result["kill_errors"] and not result["verify_errors"] + print(json.dumps(result)) + return 0 if result["passed"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/scripts/oswal-s3-bench-load.sh b/.github/scripts/oswal-s3-bench-load.sh new file mode 100644 index 00000000000..1748fb40d66 --- /dev/null +++ b/.github/scripts/oswal-s3-bench-load.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +# Runs inside the o11ybench runtime image: the load of scripts/agent-observability/greptimedb.sh +# with the same arguments plus the loader's --skip-create-table, for a table the cell created. +set -euo pipefail + +mkdir "/results/$RESULT_NAME" +mkdir -p /results/corpus +cp /work/corpus/summary.json /work/corpus/evidence.json /results/corpus/ +w="$PWD/workloads/logbench-agent-observability" +r="/results/$RESULT_NAME" +prepare_workers=${PREPARE_WORKERS:-$(python3 -c 'import sys; sys.path.insert(0, "scripts"); from benchmark_workers import effective_cpus; print(effective_cpus())')} +o11ybench-greptimedb-loader --workload logbench-agent-observability --prepare-workers "$prepare_workers" \ + --skip-create-table \ + --grpc-endpoint "$GRPC_ENDPOINT" --sql-endpoint "$ENDPOINT" \ + --database public --table agent_observations --batch-size "${GREPTIMEDB_BATCH_SIZE:-10000}" --write-mode stream --max-in-flight "${MAX_IN_FLIGHT:-1}" \ + --flush-table --request-timeout-seconds 900 \ + --schema "$w/schema.json" --physical-index-profiles "$w/physical-index-profiles.json" \ + --data /work/corpus/agent_observations.jsonl \ + --profile-summary /work/corpus/summary.json --evidence /work/corpus/evidence.json \ + --transcript-output "$r/load-transcript.jsonl" \ + --target-metadata-output "$r/target-metadata.json" \ + --greptimedb-image "$TARGET_IMAGE" --greptimedb-version "$TARGET_VERSION" \ + --greptimedb-digest "$TARGET_DIGEST" \ + --startup-input=standalone --startup-input=start \ + --startup-input=--http-addr=0.0.0.0:4000 \ + --startup-input=--grpc-bind-addr=0.0.0.0:4001 \ + > "$r/load-result.json" diff --git a/.github/scripts/oswal-s3-bench-run.sh b/.github/scripts/oswal-s3-bench-run.sh index a9835c1e5f1..f822a4a9b6f 100644 --- a/.github/scripts/oswal-s3-bench-run.sh +++ b/.github/scripts/oswal-s3-bench-run.sh @@ -22,12 +22,12 @@ check_runtime() { } pull_images() { - : "${RESULTS:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_TAG:?}" + : "${RESULTS:?}" "${RUNTIME_IMAGE:?}" "${GREPTIMEDB_REF:?}" mkdir -p "$RESULTS/images" bash "$here/pull-benchmark-image.sh" "$RUNTIME_IMAGE" - bash "$here/pull-benchmark-image.sh" "greptime/greptimedb:$GREPTIMEDB_TAG" + bash "$here/pull-benchmark-image.sh" "$GREPTIMEDB_REF" docker image inspect "$RUNTIME_IMAGE" > "$RESULTS/images/runtime.json" - docker image inspect "greptime/greptimedb:$GREPTIMEDB_TAG" > "$RESULTS/images/greptimedb.json" + docker image inspect "$GREPTIMEDB_REF" > "$RESULTS/images/greptimedb.json" } wal_disk() { @@ -46,25 +46,240 @@ wal_disk() { findmnt -no SOURCE,FSTYPE -T "$root" | tee "$RESULTS/wal-disk.txt" } +# One corpus per row count the load cells use; crash cells write their own rows. 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 + local corpus args arg dir rows + while IFS= read -r corpus; do + args=() dir="$CORPUS_DIR/$(jq -r .dir <<< "$corpus")" rows=$(jq .rows <<< "$corpus") + while IFS= read -r arg; do args+=("$arg"); done < <(jq -r '.generator_args[]' <<< "$corpus") + mkdir -p "$dir" + runtime --name "$OWNER-generator-$rows" --mount "type=bind,src=$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 "$rows" '.line_count == $rows' "$dir/summary.json" >/dev/null + done < <(jq -c '.corpora[]' "$PLAN") } -# Scrape once, after the load succeeded or failed, while the server still runs. +# Scrape once per server, after the load succeeded or failed, while it still runs: +# the loaded server into metrics.prom, a restarted one into metrics-restart.prom. 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 + grep -E '^(# (HELP|TYPE) )?(greptime_logstore_object_store_wal_|opendal_operation_(duration_seconds_count|errors_total))' \ + "$dir/metrics.all" > "$dir/metrics$phase.prom" || true rm -f "$dir/metrics.all" - metrics_scraped=true + case "$phase" in "") metrics_scraped=true ;; -restart) restart_scraped=true ;; esac +} + +# A crash cycle cannot repeat a scrape once its server is killed, so it retries. +# The named WAL sample must be in the scrape, or it does not count. +scrape_retry() { + local required=$1 attempt + for attempt in 1 2 3; do + scrape_metrics && grep -q "^greptime_logstore_object_store_wal_$required " "$dir/metrics$phase.prom" && return 0 + sleep 1 + done + return 1 +} + +sql() { curl --noproxy '*' -fsS --max-time 30 --data-urlencode "sql=$1" 'http://127.0.0.1:4000/v1/sql?db=public'; } +now_ns() { python3 -c 'import time; print(time.time_ns())'; } + +now() { date +%s.%N; } +elapsed() { awk -v a="$1" -v b="$(now)" 'BEGIN {printf "%.3f", b - a}'; } +# CPU seconds the server's cgroup used, and busy and total jiffies of the host. +server_cpu() { docker exec "$cid" cat /sys/fs/cgroup/cpu.stat 2>/dev/null | awk '/^usage_usec / {printf "%.3f", $2 / 1e6}' || true; } +host_cpu() { awk '/^cpu / {print $2 + $3 + $4 + $7 + $8 + $9, $2 + $3 + $4 + $5 + $6 + $7 + $8 + $9}' /proc/stat 2>/dev/null || true; } + +# A server's log is named after the phase it started in, whatever phase stops it. +start_server() { + server_phase=$phase + cid=$(docker run -d --init --name "$name$phase" --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$phase exited" >&2; exit 1; } + ((SECONDS < deadline)) || { echo "$id: server$phase not healthy in 300s" >&2; exit 1; } + sleep 0.2 + done +} + +# kill sends SIGKILL, killed collects a server the caller already killed, and +# no argument stops it gracefully. +stop_server() { + case "${1:-}" in + kill) docker kill --signal KILL "$cid" >/dev/null ;; + killed) docker wait "$cid" >/dev/null ;; + *) docker stop -t 120 "$cid" >/dev/null ;; + esac + docker logs "$cid" > "$dir/server$server_phase.log" 2>&1 + docker rm -fv "$cid" >/dev/null + cid= +} + +# Kill the loaded server, start the same configuration on the same prefix and +# data home, and time it until /health and until the loaded table answers. +restart() { + local table count started deadline + table=$(jq -er .table "$dir/greptimedb-1/load-result.json") + scrape_metrics || { echo "$id: metrics scrape failed" >&2; exit 4; } + # The objects recovery has to read are those still indexed at the kill. + indexed_before_kill=$(awk '/^greptime_logstore_object_store_wal_indexed_objects / {print $2}' "$dir/metrics.prom") + stop_server kill + phase=-restart started=$(now) + start_server + restart_seconds=$(elapsed "$started") + deadline=$((SECONDS + 300)) + # Regions may still be opening after /health. + until count=$(curl --noproxy '*' -fsS --max-time 30 --data-urlencode "sql=SELECT count(*) FROM $table" \ + 'http://127.0.0.1:4000/v1/sql?db=public' | jq -er '.output[0].records.rows[0][0]'); do + ((SECONDS < deadline)) || { echo "$id: $table not readable in 300s after restart" >&2; exit 5; } + sleep 0.2 + done + restart_query_seconds=$(elapsed "$started") restart_count=$count + [[ "$count" == "$(jq .rows <<< "$spec")" ]] || { echo "$id: $count rows after restart" >&2; exit 5; } + # The restart was measured, but on fewer objects than the cell asked for. + awk -v n="${indexed_before_kill:-0}" -v min="$(jq -r '.min_objects // 0' <<< "$spec")" 'BEGIN {exit !(n >= min)}' \ + || { echo "$id: ${indexed_before_kill:-no} objects indexed at the kill, fewer than min_objects" >&2; exit 7; } +} + +# Pins the WAL with one row in a second table before the writers start, if asked, +# and samples the object counts every 30 s while they run. +sample_objects() { + set +e + local body + while sleep "${OSWAL_SAMPLE_SECONDS:-30}"; do + # A failed scrape is a missing sample, never zero objects. + if ! body=$(curl --noproxy '*' -fsS --max-time 10 http://127.0.0.1:4000/metrics 2>/dev/null) \ + || ! grep -q '^greptime_logstore_object_store_wal_indexed_objects ' <<< "$body"; then + echo "$(date +%s) missing" >> "$dir/objects.tsv" + continue + fi + awk -v t="$(date +%s)" ' + /^greptime_logstore_object_store_wal_indexed_objects / {i = $2} + /^greptime_logstore_object_store_wal_deleted_objects_total / {d = $2} + /^greptime_logstore_object_store_wal_created_objects_total / {c = $2} + END {print t, i + 0, d + 0, c + 0}' <<< "$body" >> "$dir/objects.tsv" + done +} + +# Writers insert striped batches until a SIGKILL at a random point stops the +# server; it restarts on the same prefix and data home and the table is checked +# against what was acknowledged. Nothing is cleaned between cycles. +crash_cycles() { + local writers rows cycles kill_min kill_max cycle cdir wpid spid= delay before after started restart_s query_s next=0 pin_count pin_args + local load_start complete pin_left pin_base deleted sql_url=${OSWAL_SQL_URL:-http://127.0.0.1:4000/v1/sql?db=public} + writers=$(jq -r .concurrency <<< "$spec") rows=$(jq -r .batch_rows "$PLAN") cycles=$(jq -r .cycles <<< "$spec") + kill_min=$(jq -r '.kill_after[0]' <<< "$spec") kill_max=$(jq -r '.kill_after[1]' <<< "$spec") + if jq -e .pin <<< "$spec" >/dev/null; then + # Only the objects indexed before the pin row lie below it, such as the empty start object that + # the row's write lets the store collect at any later point; the servers may collect no more. + phase=-pin; scrape_retry indexed_objects || { echo "$id: metrics scrape failed" >&2; exit 4; } + read -r pin_left pin_base < <(awk '/^greptime_logstore_object_store_wal_indexed_objects / {i = $2} + /^greptime_logstore_object_store_wal_deleted_objects_total / {d = $2} END {printf "%d %d\n", i, d}' "$dir/metrics-pin.prom") + # The pin table's own auto flush outlasts every cycle a crash group allows, so its row holds the WAL. + sql "CREATE TABLE pin (v BIGINT, ts TIMESTAMP TIME INDEX) WITH ('auto_flush_interval' = '1d')" >/dev/null + sql "INSERT INTO pin VALUES (1, 0)" >/dev/null + fi + sql "CREATE TABLE crash (seq BIGINT, ts TIMESTAMP TIME INDEX) WITH (append_mode = 'true')" >/dev/null + for ((cycle = 1; cycle <= cycles; cycle++)); do + cdir="$dir/cycle-$cycle" complete=true; mkdir -p "$cdir" + # The counters of this server process before its writers start are the cycle's baseline. + phase=-start-$cycle; scrape_retry indexed_objects || complete=false + load_start=$(now_ns) + python3 "$here/oswal-s3-bench-crash.py" write --dir "$cdir" --url "$sql_url" \ + --writers "$writers" --first-seq "$next" --rows "$rows" & + wpid=$! + if jq -e .pin <<< "$spec" >/dev/null; then sample_objects & spid=$!; fi + delay=$(awk -v a="$kill_min" -v b="$kill_max" 'BEGIN {srand(); printf "%.3f", a + rand() * (b - a)}') + sleep "$delay" + phase=-cycle-$cycle; scrape_retry indexed_objects || complete=false + before=$(now_ns); docker kill --signal KILL "$cid" >/dev/null; after=$(now_ns) + wait "$wpid" + if [[ -n "$spid" ]]; then kill "$spid" 2>/dev/null || true; wait "$spid" 2>/dev/null || true; spid=; fi + stop_server killed + phase=-restart-$cycle started=$(now) + start_server + restart_s=$(elapsed "$started") + until sql "SELECT seq, count(*) FROM crash GROUP BY seq" > "$cdir/found.json" 2>/dev/null; do + (($(elapsed "$started" | cut -d. -f1) < 300)) || { echo "$id: crash not readable after restart $cycle" >&2; exit 5; } + sleep 0.2 + done + query_s=$(elapsed "$started") + scrape_retry recovery_seconds_count || complete=false + pin_count=null pin_args=() + if jq -e .pin <<< "$spec" >/dev/null; then + pin_count=$(sql "SELECT count(*) FROM pin" | jq -er '.output[0].records.rows[0][0]') pin_args=(--pin-rows "$pin_count") + fi + local verdict=0 + python3 "$here/oswal-s3-bench-crash.py" check --run-dir "$dir" --cycle "$cycle" --found "$cdir/found.json" \ + --mode "$(jq -r .wal.ACK_MODE <<< "$spec")" --writers "$writers" --rows "$rows" \ + --before-ns "$before" --after-ns "$after" --window-max-ms 2000 ${pin_args[@]+"${pin_args[@]}"} \ + > "$cdir/check.json" || verdict=$? + jq -c --arg delay "$delay" --arg restart "$restart_s" --arg query "$query_s" --argjson pin "$pin_count" \ + --argjson load "$(awk -v a="$load_start" -v b="$before" 'BEGIN {printf "%.3f", (b - a) / 1e9}')" \ + --argjson complete "$complete" \ + '. + {kill_after_seconds: ($delay | tonumber), load_seconds: $load, restart_seconds: ($restart | tonumber), + restart_query_seconds: ($query | tonumber), pin_rows: $pin, metrics_complete: $complete}' \ + "$cdir/check.json" >> "$dir/cycles.jsonl" + ((verdict == 0)) || { echo "$id: cycle $cycle failed its check" >&2; exit 6; } + # A cycle without its scrapes is not a measurement, whatever the table held. + [[ "$complete" == true ]] || { echo "$id: cycle $cycle lost a metrics scrape" >&2; exit 4; } + # Each server counts its collections from its own start: the first one from its pin scrape, + # every restarted one from zero, and an absent counter has not moved. + if jq -e .pin <<< "$spec" >/dev/null; then + deleted=$(awk '/^greptime_logstore_object_store_wal_deleted_objects_total / {d = $2} END {printf "%d", d}' \ + "$dir/metrics-cycle-$cycle.prom") + pin_left=$((pin_left - (deleted - pin_base))) pin_base=0 + ((pin_left >= 0)) || { echo "$id: cycle $cycle collected WAL objects the pin held" >&2; exit 11; } + fi + next=$(cat "$cdir"/attempted-*.log | sort -n | tail -1); next=$((next + 1)) + done +} + +# Writes the table `dropped` for write_seconds, drops it and kills the server right +# away, while most of its objects are still indexed; restarts on the same prefix and +# data home, then writes and flushes the table `kept` and samples the indexed objects +# until the dropped table's objects are collected. +drop_restart() { + local writers rows seconds started deadline indexed recovered sql_url=${OSWAL_SQL_URL:-http://127.0.0.1:4000/v1/sql?db=public} + writers=$(jq -r .concurrency <<< "$spec") rows=$(jq -r .batch_rows "$PLAN") seconds=$(jq -r .write_seconds <<< "$spec") + sql "CREATE TABLE dropped (seq BIGINT, ts TIMESTAMP TIME INDEX) WITH (append_mode = 'true')" >/dev/null + sql "CREATE TABLE kept (seq BIGINT, ts TIMESTAMP TIME INDEX) WITH (append_mode = 'true')" >/dev/null + python3 "$here/oswal-s3-bench-crash.py" write --dir "$dir/dropped" --url "$sql_url" --table dropped \ + --writers "$writers" --first-seq 0 --rows "$rows" --seconds "$seconds" \ + || { echo "$id: the writers of dropped stopped before $seconds s" >&2; exit 10; } + phase=-written; scrape_retry indexed_objects || { echo "$id: metrics scrape failed" >&2; exit 4; } + sql "DROP TABLE dropped" >/dev/null + stop_server kill + phase=-restart started=$(now) + start_server + restart_seconds=$(elapsed "$started") + scrape_retry recovery_seconds_count || { echo "$id: metrics scrape failed" >&2; exit 4; } + recovered=$(awk '/^greptime_logstore_object_store_wal_recovered_objects_total / {print $2}' "$dir/metrics-restart.prom") + python3 "$here/oswal-s3-bench-crash.py" write --dir "$dir/kept" --url "$sql_url" --table kept \ + --writers 1 --first-seq 0 --rows "$rows" --seconds 2 \ + || { echo "$id: the writer of kept stopped before 2 s" >&2; exit 10; } + sql "ADMIN FLUSH_TABLE('kept')" >/dev/null + # The flushed watermark becomes durable with the next object, so every sample writes one row. + deadline=$((SECONDS + ${OSWAL_COLLECT_SECONDS:-120})) + while :; do + sql "INSERT INTO kept VALUES (-1, $SECONDS)" >/dev/null + # The samples have their own file; the final scrape keeps the restart's name. + phase=-collected + scrape_retry indexed_objects || { phase=-restart; echo "$id: metrics scrape failed" >&2; exit 4; } + phase=-restart + indexed=$(awk '/^greptime_logstore_object_store_wal_indexed_objects / {print $2}' "$dir/metrics-collected.prom") + printf '%s %s\n' "$(date +%s)" "$indexed" >> "$dir/collected.tsv" + # A start object, the newest object and kept's own may stay; the dropped table's go. + awk -v i="$indexed" -v r="${recovered:-0}" 'BEGIN {exit !(i <= 3 || 2 * i < r)}' && break + ((SECONDS < deadline)) || { echo "$id: $indexed of ${recovered:-?} recovered objects still indexed" >&2; exit 9; } + sleep 1 + done } finish() { @@ -74,15 +289,24 @@ finish() { 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 + stop_server fi [[ -z "$wal_dir" ]] || sudo -n rm -rf "$wal_dir" + [[ -z "$home_dir" ]] || sudo -n rm -rf "$home_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" \ + --argjson restart_scraped "$restart_scraped" --arg restart_seconds "$restart_seconds" \ + --arg restart_query_seconds "$restart_query_seconds" --arg restart_count "$restart_count" \ + --arg indexed_before_kill "$indexed_before_kill" \ + --arg server_cpu_seconds "$server_cpu_seconds" --arg host_cpu_busy "$host_cpu_busy" --arg host_cpus "$(nproc 2>/dev/null)" \ '$spec + {wal_device: $device, version: $version, exit_code: $exit_code, metrics_scraped: $scraped, - load_wall_seconds: ($load_seconds | tonumber? // null)}' > "$dir/cell.json" + load_wall_seconds: ($load_seconds | tonumber? // null), + server_cpu_seconds: ($server_cpu_seconds | tonumber? // null), + host_cpu_busy: ($host_cpu_busy | tonumber? // null), host_cpus: ($host_cpus | tonumber? // null), + restart_scraped: $restart_scraped, restart_seconds: ($restart_seconds | tonumber? // null), + restart_query_seconds: ($restart_query_seconds | tonumber? // null), + restart_count: ($restart_count | tonumber? // null), + indexed_before_kill: ($indexed_before_kill | tonumber? // null)}' > "$dir/cell.json" exit "$rc" } @@ -91,22 +315,30 @@ finish() { 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 + spec=$1 cid= wal_dir= home_dir= version= load_seconds= metrics_scraped=false id= phase= server_phase= + restart_scraped=false restart_seconds= restart_query_seconds= restart_count= server_cpu_seconds= host_cpu_busy= + indexed_before_kill= + local concurrency setting started cpu_before host_before host_after corpus partitions load script id=$(jq -r .id <<< "$spec"); concurrency=$(jq -r .concurrency <<< "$spec") - dir="$RESULTS/cells/$id"; name="$OWNER-$id" + 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") + # A restarted server needs the metadata the loaded one kept in its data home. + jq -e '.restart or .mode != "load"' <<< "$spec" >/dev/null && settings+=(STORAGE__DATA_HOME=/oswal_home) while IFS= read -r setting; do settings+=("$setting"); done \ < <(jq -r '.wal | to_entries[] | "WAL__\(.key)=\(.value)"' <<< "$spec") - local docker_env=() mounts=() + docker_env=() mounts=() for setting in "${settings[@]}"; do docker_env+=(-e "GREPTIMEDB_STANDALONE__$setting") printf 'GREPTIMEDB_STANDALONE__%s\n' "$setting" >> "$dir/config.env" done + while IFS= read -r setting; do + docker_env+=(-e "$setting") + printf '%s\n' "$setting" >> "$dir/config.env" + done < <(jq -r '.env // {} | to_entries[] | "\(.key)=\(.value)"' <<< "$spec") # 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 @@ -119,33 +351,60 @@ cell() { else wal_device="s3://$S3_BUCKET/$RUN_PREFIX/$id/wal" fi + if jq -e '.restart or .mode != "load"' <<< "$spec" >/dev/null; then + home_dir="$WAL_ROOT/$id-home"; mkdir -p "$home_dir" + mounts+=(--mount "type=bind,src=$home_dir,dst=/oswal_home") + 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 + start_server 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. + if jq -e '.mode == "crash"' <<< "$spec" >/dev/null; then crash_cycles; return; fi + if jq -e '.mode == "drop"' <<< "$spec" >/dev/null; then drop_restart; return; fi + corpus="$CORPUS_DIR/corpus-$(jq -r .rows <<< "$spec")" partitions=$(jq -r '.partitions // 1' <<< "$spec") + # Only the load phase, run exactly as run-agent-observability.sh runs it, or with the + # loader writing into a table split into regions that the cell created first. + load=(scripts/agent-observability/greptimedb.sh load) script=() + if ((partitions > 1)); then + sql "$(python3 "$here/oswal-s3-bench.py" table --summary "$corpus/summary.json" --partitions "$partitions")" >/dev/null + load=(/work/oswal-s3-bench-load.sh) + script=(--mount "type=bind,src=$here/oswal-s3-bench-load.sh,dst=/work/oswal-s3-bench-load.sh,readonly") + fi + cpu_before=$(server_cpu) host_before=$(host_cpu) + started=$(now) 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 \ + --mount "type=bind,src=$corpus,dst=/work/corpus,readonly" ${script[@]+"${script[@]}"} \ + --mount "type=bind,src=$dir,dst=/results" \ + -e GRPC_ENDPOINT=127.0.0.1:4001 -e "DATA_PROFILE=$(jq -r .data_profile <<< "$spec")" -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' \ + "$RUNTIME_IMAGE" "${load[@]}" > "$dir/load.log" 2>&1 + load_seconds=$(elapsed "$started") + host_after=$(host_cpu) + server_cpu_seconds=$(awk -v a="$cpu_before" -v b="$(server_cpu)" 'BEGIN {if (a != "" && b != "") printf "%.3f", b - a}') + # Busy share of all host CPUs during the load, the loader included. + host_cpu_busy=$(awk -v a="$host_before" -v b="$host_after" \ + 'BEGIN {split(a, x, " "); split(b, y, " "); if (y[2] > x[2]) printf "%.4f", (y[1] - x[1]) / (y[2] - x[2])}') + jq -e --argjson rows "$(jq .rows <<< "$spec")" '.count_before == 0 and .count_after == $rows' \ "$dir/greptimedb-1/load-result.json" >/dev/null + if ((partitions > 1)); then + # The server's own account of the rows each region took. + sql "SELECT p.partition_name, s.region_id, s.region_rows, s.written_bytes_since_open + FROM information_schema.partitions p JOIN information_schema.region_statistics s + ON p.greptime_partition_id = s.region_id + WHERE p.table_schema = 'public' AND p.table_name = 'agent_observations' + ORDER BY s.region_id" > "$dir/regions.json" + jq -e --argjson n "$partitions" '[.output[0].records.rows[] | select(.[2] > 0)] | length == $n' \ + "$dir/regions.json" >/dev/null || { echo "$id: fewer than $partitions regions took rows" >&2; exit 8; } + fi + # An idle window: what the server sends with no writes, between two scrapes. + if jq -e '.idle_seconds > 0' <<< "$spec" >/dev/null; then + phase=-idle; scrape_metrics || { echo "$id: metrics scrape failed" >&2; exit 4; }; phase= + sleep "$(jq -r .idle_seconds <<< "$spec")" + fi + if jq -e '.restart' <<< "$spec" >/dev/null; then restart; fi } scrub() { diff --git a/.github/scripts/oswal-s3-bench-summary.py b/.github/scripts/oswal-s3-bench-summary.py index 67186d422da..305a647a9a4 100644 --- a/.github/scripts/oswal-s3-bench-summary.py +++ b/.github/scripts/oswal-s3-bench-summary.py @@ -26,14 +26,30 @@ 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") +# The Raft Engine controls every object store WAL cell is compared with: fsync per write and the default. +BASELINES = ("R", "R0") SAMPLE = re.compile(r'([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+(\S+)') +# Mito's line for every region it opened, with the WAL range it replayed. +REPLAY = re.compile(r"Replay WAL for region: (\d+)\(.*?provider: (\w+).*?rows recovered: (\d+), " + r"replay from entry id: (\d+), last entry id: (\d+),.*?elapsed: ([0-9.]+)(ns|µs|us|ms|s)\b") +UNITS = {"ns": 1e-9, "µs": 1e-6, "us": 1e-6, "ms": 1e-3, "s": 1.0} +# An object store WAL entry id is the object sequence above a 20-bit position. +POSITION_BITS = 20 def parse_metrics(text: str) -> dict: - """Map each unlabeled sample to its value and each histogram to its buckets.""" - samples, buckets = {}, {} + """Map each unlabeled sample to its value, each histogram to its buckets, and the S3 + operations (data files and WAL together, as OpenDAL counts its calls: a paginated list or a + multipart write is one operation over several HTTP requests) to their count per kind; the local cache + stores report through the same layer under another scheme and are left out.""" + samples, buckets, operations = {}, {}, {} for line in text.splitlines(): match = SAMPLE.fullmatch(line.strip()) + if match and match.group(1) == "opendal_operation_duration_seconds_count": + operation = re.search(r'operation="([^"]+)"', match.group(2) or "") + if operation and re.search(r'scheme="s3"', match.group(2) or ""): + operations[operation.group(1)] = operations.get(operation.group(1), 0) + float(match.group(3)) + continue 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)) @@ -42,7 +58,8 @@ def parse_metrics(text: str) -> dict: 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()}} + return {"samples": samples, "buckets": {name: sorted(items) for name, items in buckets.items()}, + "operations": operations} def quantile(q: float, buckets: list) -> float | None: @@ -64,25 +81,149 @@ 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 read_metrics(path: Path) -> dict: + return parse_metrics(path.read_text() if path.exists() else "") + + +def ratio(a, b): + return a / b if a is not None and b else None + + +def replay(log: Path, metrics: dict) -> dict | None: + """What a restarted server replayed: every region it opened, the span of object sequences + its WAL range covers, and the log store's time serving region reads (`read_seconds`). The span + bounds the objects a region read from above: objects without its entries are skipped.""" + if not log.exists(): + return None + regions = [] + for match in REPLAY.finditer(log.read_text(errors="replace")): + region, provider, rows, first, last, elapsed, unit = match.groups() + # The range runs from the first entry to replay to the last one; it is empty when first > last. + span = None if provider != "ObjectStore" else 0 if int(first) > int(last) else \ + (int(last) >> POSITION_BITS) - (int(first) >> POSITION_BITS) + 1 + regions.append({"region": int(region), "rows": int(rows), "first_entry_id": int(first), + "last_entry_id": int(last), "seconds": float(elapsed) * UNITS[unit], "object_span": span}) + samples = metrics["samples"] + return {"regions": regions, "rows": sum(r["rows"] for r in regions), + "seconds": max((r["seconds"] for r in regions), default=None), + "object_span": sum(r["object_span"] for r in regions) + if regions and regions[0]["object_span"] is not None else None, + "read_seconds": samples.get("read_seconds_sum"), "reads": samples.get("read_seconds_count")} + + +def replayed(regions: list) -> str: + """Each region that replayed rows: its rows, the object sequences its range spans and its time.""" + return "; ".join(f"{r['region']}: {r['rows']:,} rows" + ( + f", object sequences {r['first_entry_id'] >> POSITION_BITS:,}–{r['last_entry_id'] >> POSITION_BITS:,}" + if r["object_span"] else "") + f", {r['seconds']:.2f} s" for r in regions if r["rows"]) or "—" + + +def client_latency(path: Path) -> tuple: + """The 50th and 99th percentile of the loader's own time per write request.""" + durations = [] + if path.exists(): + for line in path.read_text().splitlines(): + record = json.loads(line) if line.strip() else {} + if record.get("step") == "grpc_write_batch" and record.get("status") == "success": + durations.append(record["duration_seconds"]) + durations.sort() + return tuple(durations[max(0, math.ceil(q * len(durations)) - 1)] if durations else None for q in (0.5, 0.99)) + + +def region_rows(path: Path) -> list | None: + """Rows per region of a partitioned load cell, as the server reported them after the load.""" + if not path.exists(): + return None + return [int(row[2] or 0) for row in json.loads(path.read_text())["output"][0]["records"]["rows"]] + + +def interval_seconds(wal: dict) -> float: + value = wal.get("FLUSH_INTERVAL", "100ms") + return float(value[:-2]) / 1000 if value.endswith("ms") else float(value.rstrip("s")) + + +def traffic(end: dict, start: dict, seconds, wal: dict) -> dict: + """WAL traffic between two scrapes of one process over `seconds`; more objects than seal + ticks, or objects near the 8 MiB threshold, mean batches sealed on size rather than on the timer.""" + def delta(name): + value = end["samples"].get(name) + return None if value is None else value - start["samples"].get(name, 0) + created, written, objects = delta("created_objects_total"), delta("object_bytes_sum"), delta("object_bytes_count") + per_second = created / seconds if created and seconds else None + mean = written / objects if written and objects else None + return {"wal_mb_per_second": written / seconds / 1e6 if written and seconds else None, + "objects_per_second": per_second, "mean_object_bytes": mean, + "sealed_on_size": None if per_second is None else + per_second > 1.05 / interval_seconds(wal) or (mean or 0) >= 0.9 * 8 * 2**20, + "operations": {op: n - start["operations"].get(op, 0) for op, n in end["operations"].items() + if n != start["operations"].get(op, 0)}} + + +def crash_cycles(path: Path, wal: dict) -> list: + """Each cycle's check plus the loaded server's metrics before the kill and the restarted one's after.""" + cycles = [json.loads(line) for line in (path / "cycles.jsonl").read_text().splitlines() + if line.strip()] if (path / "cycles.jsonl").exists() else [] + for cycle in cycles: + start = read_metrics(path / f"metrics-start-{cycle['cycle']}.prom") + killed = read_metrics(path / f"metrics-cycle-{cycle['cycle']}.prom") + before = killed["samples"] + after = read_metrics(path / f"metrics-restart-{cycle['cycle']}.prom") + once = after["samples"].get("recovery_seconds_count") == 1 + cycle |= traffic(killed, start, cycle.get("load_seconds"), wal) + cycle |= {"created_before_kill": before.get("created_objects_total"), + "indexed_before_kill": before.get("indexed_objects"), + "recovery_seconds": after["samples"].get("recovery_seconds_sum") if once else None, + "recovered_objects": after["samples"].get("recovered_objects_total", 0 if once else None), + "restart_operations": after["operations"], + "replay": replay(path / f"server-restart-{cycle['cycle']}.log", after)} + return cycles + + 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", []) + final, restarted = read_metrics(path / "metrics.prom"), read_metrics(path / "metrics-restart.prom") + # With an idle window the scrape taken when the load ended describes the load. + metrics = read_metrics(path / "metrics-idle.prom") if cell.get("idle_seconds") else final load_seconds = result.get("load_seconds") + wall, cpu = cell.get("load_wall_seconds"), cell.get("server_cpu_seconds") row = { - "id": cell["id"], "round": cell["round"], "concurrency": cell["concurrency"], + "id": cell["id"], "group": cell.get("group", ""), "mode": cell.get("mode", "load"), + "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"), + "missing": cell.get("missing", False), "env": cell.get("env", {}), + "version": cell.get("version"), "load_wall_seconds": wall, "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), + "server_cpu_cores": cpu / wall if cpu is not None and wall else None, + "host_cpu_busy": cell.get("host_cpu_busy"), "host_cpus": cell.get("host_cpus"), } + for name, key in (("append_ack_seconds", "append_ack"), ("seal_to_durable_seconds", "seal_to_durable"), + ("object_bytes", "object_bytes")): + buckets = metrics["buckets"].get(name, []) + row[f"{key}_p50"], row[f"{key}_p99"] = quantile(0.5, buckets), quantile(0.99, buckets) + row["client_p50"], row["client_p99"] = client_latency(path / "greptimedb-1" / "load-transcript.jsonl") + # An entry is one region's write; an append, counted only in the durable mode, carries one or more. + samples = metrics["samples"] + row["entries_per_object"] = ratio(samples.get("object_entries_sum"), samples.get("object_entries_count")) + row["appends_per_object"] = ratio(samples.get("append_ack_seconds_count"), samples.get("object_bytes_count")) + row["partitions"] = cell.get("partitions", 1) + row["region_rows"] = region_rows(path / "regions.json") + if cell.get("restart"): + # A restarted process observed exactly one recovery, so the sum is its duration. + count = restarted["samples"].get("recovery_seconds_count") + row["restart"] = { + "wall_seconds": cell.get("restart_seconds"), "query_seconds": cell.get("restart_query_seconds"), + "rows": cell.get("restart_count"), + "recovery_seconds": restarted["samples"].get("recovery_seconds_sum") if count == 1 else None, + "recovered_objects": restarted["samples"].get("recovered_objects_total") + if cell.get("restart_scraped") else None, + "indexed_before_kill": cell.get("indexed_before_kill"), "min_objects": cell.get("min_objects", 0), + "replay": replay(path / "server-restart.log", restarted), + } # 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" @@ -91,26 +232,74 @@ def load_cell(path: Path, spec: dict | None = None) -> dict: 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 + row |= traffic(metrics, {"samples": {}, "operations": {}}, load_seconds, cell.get("wal", {})) + row["restart_operations"] = restarted["operations"] + if cell.get("idle_seconds"): + idle = traffic(final, metrics, cell["idle_seconds"], cell.get("wal", {})) + end = final["samples"].get("created_objects_total") + row["idle"] = {"seconds": cell["idle_seconds"], "operations": idle["operations"], + "creates": None if end is None else end - metrics["samples"].get("created_objects_total", 0)} + if row["mode"] == "drop": + written = read_metrics(path / "metrics-written.prom")["samples"] + restarted_samples = restarted["samples"] + samples = [line.split() for line in (path / "collected.tsv").read_text().splitlines() + if line.strip()] if (path / "collected.tsv").exists() else [] + row["drop"] = { + "acked_batches": sum(len((p.read_text()).split()) for p in sorted((path / "dropped").glob("acked-*.log"))), + "indexed_before_drop": written.get("indexed_objects"), + "restart_seconds": cell.get("restart_seconds"), + "recovered_objects": restarted_samples.get("recovered_objects_total"), + "indexed_at_end": int(float(samples[-1][1])) if samples else None, + "deleted_after_restart": restarted_samples.get("deleted_objects_total"), + "seconds_to_collect": int(samples[-1][0]) - int(samples[0][0]) if samples else None, + } + if row["mode"] == "crash": + row["cycles"] = crash_cycles(path, cell.get("wal", {})) + if (path / "objects.tsv").exists(): + # A sample is time, indexed, deleted, created; a failed scrape is time and "missing". + row["objects_samples"] = [[int(float(line.split()[0]))] + ( + [None] * 3 if line.split()[1:] == ["missing"] else [int(float(v)) for v in line.split()[1:]]) + for line in (path / "objects.tsv").read_text().splitlines() if line.strip()] 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} + """Each object store WAL cell over each Raft Engine control per round, and their geometric mean over the rounds, per concurrency.""" + rows = [r for r in rows if r["mode"] == "load"] + speed = {(r["group"], 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}) + for group in dict.fromkeys(r["group"] for r in rows): + members = [r for r in rows if r["group"] == group] + series = sorted({r["config"] for r in members if r["config"] not in BASELINES}) + baselines = [b for b in BASELINES if any(r["config"] == b for r in members)] + for concurrency in sorted({r["concurrency"] for r in members}): + rounds = sorted({r["round"] for r in members if r["concurrency"] == concurrency}) + for config, baseline in ((s, b) for s in series for b in baselines): + per_round = [] + for round_number in rounds: + base = speed.get((group, round_number, concurrency, baseline)) + measured = speed.get((group, round_number, concurrency, config)) + per_round.append(measured / base if base and measured else None) + complete = [ratio for ratio in per_round if ratio is not None] + result.append({"group": group, "concurrency": concurrency, "ratio": f"{config}/{baseline}", + "per_round": per_round, + "geomean": geomean(complete) if len(complete) == len(per_round) else None}) return result +def operations(counts: dict) -> str: + return ", ".join(f"{op} {n:,.0f}" for op, n in sorted(counts.items())) or "—" + + +def overrides(env: dict) -> str: + return " ".join(f"{k}={v}" for k, v in sorted(env.items())) or "—" + + +def pair(a, b, digits=0, scale=1.0) -> str: + return f"{fmt(a, digits, scale)} / {fmt(b, digits, scale)}" + + def fmt(value, digits=0, scale=1.0) -> str: return "—" if value is None else f"{value * scale:,.{digits}f}" @@ -119,21 +308,134 @@ 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 \ + lines += ["| Cell | Overrides | Rows/s | Load s | Wall s | Ack p50 / p99 ms | Seal to durable p50 / p99 ms " + "| Object p50 / p99 MiB | Creates | Creates/min | Stalled | Poisoned | Create failures | Conflicts " + "| Server CPUs | Host busy | WAL on |", + "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: " + "| ---: | ---: | --- |"] + def label(r): + return 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']})" + loads = [r for r in rows if r["mode"] == "load"] + for r in loads: + cell = label(r) 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), + f"| {cell}", overrides(r["env"]), fmt(r["rows_per_second"]), fmt(r["load_seconds"], 1), + fmt(r["load_wall_seconds"], 1), pair(r["append_ack_p50"], r["append_ack_p99"], 1, 1000), + pair(r["seal_to_durable_p50"], r["seal_to_durable_p99"], 1, 1000), + pair(r["object_bytes_p50"], r["object_bytes_p99"], 1, 1 / 2**20), *(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 |", "| ---: | --- | --- | ---: |"] + *(fmt(r[c]) for c in COUNTERS[1:]), fmt(r["server_cpu_cores"], 2), + "—" if r["host_cpu_busy"] is None else f"{r['host_cpu_busy']:.0%} of {r['host_cpus'] or '?'}", + f"`{r['wal_device'] or '—'}` |"])) + lines += ["", "| Cell | WAL MB/s | Objects/s | Mean object MiB | Sealed on size | OpenDAL S3 operations during the load " + "| OpenDAL S3 operations of the restart | Idle s | Idle creates | Idle OpenDAL S3 operations |", + "| --- | ---: | ---: | ---: | --- | --- | --- | ---: | ---: | --- |"] + for r in loads: + idle = r.get("idle") or {} + lines.append(" | ".join([ + f"| {label(r)}", fmt(r["wal_mb_per_second"], 1), fmt(r["objects_per_second"], 1), + fmt(r["mean_object_bytes"], 2, 1 / 2**20), + "—" if r["sealed_on_size"] is None else ("yes" if r["sealed_on_size"] else "no"), + operations(r["operations"]), operations(r["restart_operations"]) if "restart" in r else "—", + fmt(idle.get("seconds")), fmt(idle.get("creates")), operations(idle["operations"]) if idle else "—"]) + " |") + crashes = [r for r in rows if r["mode"] == "crash"] + if crashes: + lines += ["", "| Cell | Cycle | Kill after s | Acked | Present | Lost | Lost rows | Extra | In flight " + "| All found | All acked | All lost | Indexed at kill | Restart to /health s | Recovery s " + "| Recovered objects | Pin rows | Verdict |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: " + "| ---: | ---: | ---: | --- |"] + for r in crashes: + for c in r["cycles"] or [{}]: + verdict = "—" if not c else "FAIL: " + "; ".join(c.get("kill_errors", []) + c.get("verify_errors", [])) \ + if not c.get("passed") else "pass" if c.get("metrics_complete", True) else "pass, scrape missing" + lines.append(" | ".join([ + f"| {label(r)}", fmt(c.get("cycle")), fmt(c.get("kill_after_seconds"), 1), + fmt(c.get("cycle_acked_batches")), fmt(c.get("cycle_present_batches")), + fmt(c.get("cycle_lost_batches")), fmt(c.get("cycle_lost_rows")), fmt(c.get("cycle_extra_batches")), + fmt(c.get("in_flight")), fmt(c.get("total_found_batches")), fmt(c.get("total_acked_batches")), + fmt(c.get("total_lost_batches")), fmt(c.get("indexed_before_kill")), + fmt(c.get("restart_seconds"), 2), fmt(c.get("recovery_seconds"), 3), + fmt(c.get("recovered_objects")), fmt(c.get("pin_rows")), verdict]) + " |") + lines += ["", "| Cell | Cycle | Load s | WAL MB/s | Objects/s | Mean object MiB | Sealed on size " + "| OpenDAL S3 operations until the kill | OpenDAL S3 operations of the restart |", + "| --- | ---: | ---: | ---: | ---: | ---: | --- | --- | --- |"] + for r in crashes: + for c in r["cycles"]: + lines.append(" | ".join([ + f"| {r['id']}", fmt(c["cycle"]), fmt(c.get("load_seconds"), 1), fmt(c["wal_mb_per_second"], 1), + fmt(c["objects_per_second"], 1), fmt(c["mean_object_bytes"], 2, 1 / 2**20), + "—" if c["sealed_on_size"] is None else ("yes" if c["sealed_on_size"] else "no"), + operations(c["operations"]), operations(c["restart_operations"])]) + " |") + pinned = [r for r in crashes if r.get("objects_samples")] + if pinned: + lines += ["", "| Cell | Indexed objects every 30 s | Missing samples | Deleted objects at the end " + "| Created objects at the end |", "| --- | --- | ---: | ---: | ---: |"] + for r in pinned: + samples = r["objects_samples"] + last = next((s for s in reversed(samples) if s[1] is not None), [None] * 4) + lines.append(f"| {r['id']} | {', '.join('missing' if s[1] is None else str(s[1]) for s in samples)} " + f"| {sum(s[1] is None for s in samples)} | {fmt(last[2])} | {fmt(last[3])} |") + details = [r for r in loads if any(r[k] is not None for k in ("client_p50", "entries_per_object", "region_rows"))] + if details: + lines += ["", "| Cell | Client p50 / p99 ms | Entries per object | Appends per object | Regions " + "| Rows per region min / max |", "| --- | ---: | ---: | ---: | ---: | ---: |"] + for r in details: + spread = r["region_rows"] or [] + lines.append(" | ".join([ + f"| {label(r)}", pair(r["client_p50"], r["client_p99"], 1, 1000), fmt(r["entries_per_object"], 1), + fmt(r["appends_per_object"], 2), fmt(len(spread) if spread else r["partitions"]), + pair(min(spread), max(spread)) if spread else "—"]) + " |") + replays = [(r["id"], "—", r["restart"]) for r in loads if r.get("restart", {}).get("replay")] + replays += [(r["id"], fmt(c["cycle"]), c | {"wall_seconds": c.get("restart_seconds")}) + for r in crashes for c in r["cycles"] if c.get("replay")] + if replays: + lines += ["", "| Cell | Cycle | Restart to /health s | Recovery s | Recovered objects | Regions opened " + "| Rows replayed | Object span | Slowest region replay s | Read seconds (reads) " + "| Regions with rows |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- | --- |"] + for cell, cycle, x in replays: + y = x["replay"] + lines.append(" | ".join([ + f"| {cell}", cycle, fmt(x.get("wall_seconds"), 2), fmt(x.get("recovery_seconds"), 3), + fmt(x.get("recovered_objects")), fmt(len(y["regions"])), fmt(y["rows"]), fmt(y["object_span"]), + fmt(y["seconds"], 2), f"{fmt(y['read_seconds'], 2)} ({fmt(y['reads'])})", + replayed(y["regions"])]) + " |") + drops = [r for r in rows if r.get("drop")] + if drops: + lines += ["", "| Cell | Acked batches before the drop | Indexed before the drop | Restart to /health s " + "| Recovered objects | Indexed at the end | Deleted after the restart | Seconds to collect | Verdict |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |"] + for r in drops: + x = r["drop"] + # Objects the release already collected before the kill leave nothing to show. + verdict = "—" if x["indexed_at_end"] is None else \ + "nothing left at the kill" if (x["recovered_objects"] or 0) <= 3 else \ + "collected" if r["exit_code"] == 0 else "still indexed" + lines.append(" | ".join([ + f"| {label(r)}", fmt(x["acked_batches"]), fmt(x["indexed_before_drop"]), fmt(x["restart_seconds"], 2), + fmt(x["recovered_objects"]), fmt(x["indexed_at_end"]), + fmt(x["deleted_after_restart"]), fmt(x["seconds_to_collect"]), verdict]) + " |") + restarts = [r for r in rows if "restart" in r] + if restarts: + lines += ["", "| Cell | Overrides | Objects created | Indexed at kill | Restart to /health s | Restart to count s " + "| Recovery s | Recovered objects | Rows after restart |", + "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"] + for r in restarts: + x = r["restart"] + short = x["min_objects"] and (x["indexed_before_kill"] or 0) < x["min_objects"] + lines.append(" | ".join([ + f"| {label(r)}", overrides(r["env"]), fmt(r["created_objects_total"]), + fmt(x["indexed_before_kill"]) + (f" (under {x['min_objects']:,})" if short else ""), + fmt(x["wall_seconds"], 2), + fmt(x["query_seconds"], 2), fmt(x["recovery_seconds"], 3), fmt(x["recovered_objects"]), + f"{fmt(x['rows'])} |"])) + lines += ["", "| Group | 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)} |") + lines.append(f"| {p['group'] or '—'} | {p['concurrency']} | {p['ratio']} " + f"| {', '.join(fmt(v, 3) for v in p['per_round'])} | {fmt(p['geomean'], 3)} |") return "\n".join(lines) + "\n" diff --git a/.github/scripts/oswal-s3-bench.py b/.github/scripts/oswal-s3-bench.py index faad8ace674..9d674400fe2 100644 --- a/.github/scripts/oswal-s3-bench.py +++ b/.github/scripts/oswal-s3-bench.py @@ -29,18 +29,62 @@ import re import sys import time import urllib.request +from datetime import datetime 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. +# Identical [storage] for every configuration; only [wal] differs. R0 is the standalone default. WAL_CONFIGS = { "R": {"PROVIDER": "raft_engine", "SYNC_WRITE": "true", "DIR": "/wal"}, + "R0": {"PROVIDER": "raft_engine", "SYNC_WRITE": "false", "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")) +BASELINES = ("R", "R0") +# Restart cells seal ten times as often as the default so recovery has thousands of objects. +RESTART_FLUSH_INTERVAL = "10ms" +# The loader's CREATE TABLE for the agent observability workload at the pinned runtime image. +AGENT_TABLE = """CREATE TABLE `agent_observations` ( + `event_time` TIMESTAMP(3) NOT NULL, + `biz_date` DATE NOT NULL, + `trace_id` STRING NOT NULL SKIPPING INDEX, + `session_id` STRING NOT NULL, + `observation_id` STRING NOT NULL SKIPPING INDEX, + `parent_observation_id` STRING NULL SKIPPING INDEX, + `seq_no` BIGINT NOT NULL, + `type` STRING NOT NULL INVERTED INDEX, + `status` STRING NOT NULL INVERTED INDEX, + `tenant` STRING NOT NULL, + `app` STRING NOT NULL, + `environment` STRING NOT NULL, + `task_category` STRING NULL, + `trace_archetype` STRING NULL, + `model` STRING NULL, + `tool_name` STRING NULL, + `input` STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', backend = 'bloom', case_sensitive = 'false', granularity = '10240', false_positive_rate = '0.01'), + `output` STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', backend = 'bloom', case_sensitive = 'false', granularity = '10240', false_positive_rate = '0.01'), + `input_tokens` BIGINT NULL, + `output_tokens` BIGINT NULL, + `total_cost` DOUBLE NULL, + `latency_ms` BIGINT NULL, + `payload` JSON2 NOT NULL, + TIME INDEX (`event_time`), + PRIMARY KEY (`tenant`, `app`, `environment`) +) +{partitions}ENGINE=mito +WITH( + append_mode = 'true', + 'compaction.type' = 'twcs', + 'compaction.twcs.time_window' = '2h' +)""" +GROUP_NAME = re.compile(r"[a-z][a-z0-9]{0,11}") +RESERVED_ENV = ("GREPTIMEDB_STANDALONE__STORAGE__", "GREPTIMEDB_STANDALONE__WAL__") +ENV_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") +# Docker's reference grammar: path components separated by ".", "_", "__" or runs of "-". +IMAGE = re.compile(r"[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*(?::[0-9]+)?(?:/[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*)+" + r"(?::[A-Za-z0-9_][A-Za-z0-9_.-]{0,127})?(?:@sha256:[0-9a-f]{64})?") 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}") @@ -78,32 +122,155 @@ def build_manifest(plan: dict, greptimedb_inspect: list, runtime_image: str, 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"], + "configs": plan["configs"], "extra_env": plan["extra_env"], + "restart": plan["restart"], "groups": plan["groups"], "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: +def parse_extra_env(text: str) -> dict: + """A JSON object or K=V lines; the storage and WAL settings that define a cell stay the benchmark's own.""" + if text.strip().startswith("{"): + parsed = json.loads(text) + if not all(isinstance(v, (str, int)) and not isinstance(v, bool) for v in parsed.values()): + raise ValueError("extra_env values must be strings or integers") + env = {k: str(v) for k, v in parsed.items()} + else: + env = {} + for line in filter(None, (line.strip() for line in text.splitlines())): + name, sep, value = line.partition("=") + if not sep or name.strip() in env: + raise ValueError(f"extra_env line is not a new K=V: {line!r}") + env[name.strip()] = value.strip() + for name, value in env.items(): + # GreptimeDB matches its environment keys without regard to case. + if not ENV_NAME.fullmatch(name) or name.upper().startswith(RESERVED_ENV) or "\n" in value: + raise ValueError(f"extra_env may not set {name!r}") + return env + + +def partitioned_table(summary: dict, partitions: int) -> str: + """The loader's table split into equal event_time ranges over the corpus window. The + generator draws event times uniformly from that window, so every range gets an equal + share of the rows and every batch writes to every region.""" + def millis(text): + return int(datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp() * 1000) + start, end = millis(summary["start"]), millis(summary["end"]) + bounds = [start + (end - start) * i // partitions for i in range(1, partitions)] + if len(set(bounds)) != len(bounds) or not start < bounds[0]: + raise ValueError(f"the corpus window {summary['start']}..{summary['end']} is too short for {partitions} ranges") + ranges = [f"`event_time` < {bounds[0]}"] + ranges += [f"`event_time` >= {low} AND `event_time` < {high}" for low, high in zip(bounds, bounds[1:])] + ranges.append(f"`event_time` >= {bounds[-1]}") + clause = "PARTITION ON COLUMNS (`event_time`) (\n " + ",\n ".join(ranges) + "\n)\n" + return AGENT_TABLE.format(partitions=clause) + + +def group_cells(group: dict, smoke: bool) -> list: + """The cells of one group, in run order; a group without a name is the top-level inputs.""" + name, mode = group.get("name", ""), group.get("mode", "load") + if mode not in ("load", "crash", "drop"): + raise ValueError(f"group {name!r}: mode must be load, crash or drop, got {mode!r}") + concurrency = str(group["concurrency"]) 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) + chosen = [config.strip() for config in str(group["configs"]).split(",")] + allowed = set(WAL_CONFIGS) - (set(BASELINES) if mode != "load" else set()) + if not chosen or len(set(chosen)) != len(chosen) or not set(chosen) <= allowed: + raise ValueError(f"group {name!r}: configs must be distinct names among {', '.join(sorted(allowed))}, " + f"got {group['configs']!r}") + extra = group.get("extra_env", "") + env = parse_extra_env(json.dumps(extra) if isinstance(extra, dict) else str(extra)) + restart = group.get("restart", False) is True + rows = SMOKE_ROWS if smoke else positive("rows", str(group["rows"]), 100_000_000) + rounds = 1 if smoke else positive("rounds", str(group["rounds"]), 10) + idle = int(group.get("idle_seconds", 0)) + common = {"group": name, "mode": mode, "restart": restart, "idle_seconds": idle} + partitions = int(group.get("partitions", 1)) + if partitions != 1 and (mode != "load" or not 2 <= partitions <= 64): + raise ValueError(f"group {name!r}: partitions must be in 2..64 and needs mode load") + if mode == "load": + if not 0 <= idle <= 600: + raise ValueError(f"group {name!r}: idle_seconds must be in 0..600") + # The objects a restart cell must leave indexed at its kill; a smoke run is too small for any. + min_objects = int(group.get("min_objects", 0)) + if min_objects < 0 or (min_objects and not restart): + raise ValueError(f"group {name!r}: min_objects needs restart and must not be negative") + common |= {"rows": rows, "data_profile": DATA_PROFILES.get(rows, "custom"), + "min_objects": 0 if smoke else min_objects, "partitions": partitions} + elif mode == "drop": + # Writes one table for write_seconds, drops it, kills and restarts the server, then + # writes and flushes another table and watches the dropped table's objects go. + write_seconds = int(group.get("write_seconds", 30)) + if not 1 <= write_seconds <= 600 or restart or idle: + raise ValueError(f"group {name!r}: write_seconds must be in 1..600, without restart or idle") + common |= {"write_seconds": min(write_seconds, 10) if smoke else write_seconds} + else: + cycles = int(group.get("cycles", 5)) + kill_after = [int(v) for v in group.get("kill_after", [3, 10])] + if not 1 <= cycles <= 20 or len(kill_after) != 2 or not 1 <= kill_after[0] <= kill_after[1] <= 3600: + raise ValueError(f"group {name!r}: cycles must be in 1..20 and kill_after a [min, max] in 1..3600") + if restart or idle: + raise ValueError(f"group {name!r}: a crash group restarts after every kill and has no idle window") + if smoke: + cycles, kill_after = min(cycles, 2), [min(v, 60) for v in kill_after] + common |= {"cycles": cycles, "kill_after": kill_after, "pin": group.get("pin", False) is True} + series = [config for config in WAL_CONFIGS if config in chosen] cells = [] - for round_number in range(1, round_count + 1): - # Alternate the order between rounds so no configuration always runs first. + for round_number in range(1, rounds + 1): + # Reverse the order every other round 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], + for config in series if round_number % 2 else reversed(series): + wal = WAL_CONFIGS[config] | ({"FLUSH_INTERVAL": RESTART_FLUSH_INTERVAL} if restart and config not in BASELINES else {}) + cells.append(common | { + "id": (f"{name}-" if name else "") + f"{round_number}-{level}-{config}", + "round": round_number, "concurrency": level, "config": config, "wal": wal, "env": env, }) + return cells + + +def parse_groups(text: str, defaults: dict) -> list: + """A JSON list of groups; every key a group leaves out takes the top-level input's value.""" + if not text.strip(): + return [defaults] + groups = json.loads(text) + if not isinstance(groups, list) or not groups or not all(isinstance(g, dict) for g in groups): + raise ValueError("groups must be a non-empty JSON list of objects") + known = set(defaults) | {"name", "mode", "idle_seconds", "cycles", "kill_after", "pin", "min_objects", + "partitions", "write_seconds"} + names = [g.get("name") for g in groups] + for group, name in zip(groups, names): + if not isinstance(name, str) or not GROUP_NAME.fullmatch(name): + raise ValueError(f"every group needs a name of up to 12 lowercase letters and digits, got {name!r}") + if set(group) - known: + raise ValueError(f"group {name!r} has unknown keys {sorted(set(group) - known)}") + if len(set(names)) != len(names): + raise ValueError(f"group names repeat: {names}") + return [defaults | group for group in groups] + + +def build_plan(concurrency: str, rows: str, batch_rows: str, rounds: str, smoke: bool, + configs: str = "R,D,E", extra_env: str = "", restart: bool = False, groups: str = "") -> dict: + defaults = {"concurrency": concurrency, "rows": rows, "rounds": rounds, "configs": configs, + "extra_env": extra_env, "restart": restart} + resolved = parse_groups(groups, defaults) + cells = [cell for group in resolved for cell in group_cells(group, smoke)] + if len({cell["id"] for cell in cells}) != len(cells): + raise ValueError("two cells share an id") + corpora = [] + for count in dict.fromkeys(cell["rows"] for cell in cells if cell["mode"] == "load"): + profile = DATA_PROFILES.get(count) + corpora.append({"rows": count, "dir": f"corpus-{count}", "data_profile": profile or "custom", + "generator_args": ["--data-profile", profile] if profile else ["--line-count", str(count)]}) + first = cells[0] 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, + "smoke": smoke, "rows": first.get("rows"), "rounds": max(cell["round"] for cell in cells), + "batch_rows": positive("batch_rows", batch_rows, 100_000), + "concurrency": sorted({cell["concurrency"] for cell in cells}), + "configs": list(dict.fromkeys(cell["config"] for cell in cells)), + "extra_env": parse_extra_env(extra_env), "groups": [g.get("name", "") for g in resolved], + "restart": restart, "corpora": corpora, "cells": cells, } @@ -114,6 +281,15 @@ def newest_nightly(tags: list[str]) -> str: return max(dated)[1] +def resolve_image(image: str, tag: str) -> str: + """The full image reference when given, else greptime/greptimedb at the resolved tag.""" + if image: + if not IMAGE.fullmatch(image) or not re.search(r"(:[^/]+|@sha256:[0-9a-f]{64})$", image): + raise ValueError(f"greptimedb_image must be a full reference with a tag or digest: {image!r}") + return image + return f"greptime/greptimedb:{resolve_tag(tag)}" + + def resolve_tag(requested: str) -> str: if requested != "nightly": if not TAG.fullmatch(requested): @@ -169,12 +345,21 @@ 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("--restart", choices=("true", "false"), default="false") + plan.add_argument("--configs", default="R,D,E") + for name in ("extra-env", "groups"): + plan.add_argument(f"--{name}", default="") for name in ("timeout-minutes", "ttl-hours", "output"): plan.add_argument(f"--{name}") - commands.add_parser("tag").add_argument("requested") + image = commands.add_parser("image") + image.add_argument("--tag", required=True) + image.add_argument("--image", default="") regions = commands.add_parser("regions") for name in ("bucket", "runner"): regions.add_argument(f"--{name}", required=True) + table = commands.add_parser("table") + table.add_argument("--summary", type=Path, required=True) + table.add_argument("--partitions", type=int, required=True) manifest = commands.add_parser("manifest") for name in ("results", "runtime-image", "instance-type", "region", "prefix"): manifest.add_argument(f"--{name}", required=True) @@ -187,18 +372,21 @@ def main() -> int: 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") + result = build_plan(args.concurrency, args.rows, args.batch_rows, args.rounds, args.smoke == "true", + args.configs, args.extra_env, args.restart == "true", args.groups) 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}") + print(f"{len(result['cells'])} cells; corpora {[c['rows'] for c in result['corpora']]}") + elif args.command == "image": + reference = resolve_image(args.image, args.tag) + append_github("GITHUB_OUTPUT", "greptimedb_ref", reference) + print(f"GreptimeDB image: {reference}") elif args.command == "regions": check_regions(args.bucket, args.runner) print(f"bucket and runner both in {args.runner}") + elif args.command == "table": + print(partitioned_table(json.loads(args.summary.read_text()), args.partitions)) elif args.command == "manifest": results = Path(args.results) manifest = build_manifest(json.loads((results / "plan.json").read_text()), diff --git a/.github/workflows/oswal-s3-bench.yml b/.github/workflows/oswal-s3-bench.yml index c31dfe8f1c3..f5e92d2cad5 100644 --- a/.github/workflows/oswal-s3-bench.yml +++ b/.github/workflows/oswal-s3-bench.yml @@ -10,6 +10,26 @@ on: 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 + greptimedb_image: + description: 'Full image reference with a tag or digest; used instead of greptimedb_tag when set' + type: string + default: '' + configs: + description: 'WAL configurations: R (Raft Engine, sync_write), R0 (Raft Engine, standalone default), D (durable), E (enqueued)' + type: string + default: R,D,E + extra_env: + description: 'Environment for every GreptimeDB container: a JSON object or K=V lines' + type: string + default: '' + restart: + description: 'After each load, SIGKILL the server, restart it on the same prefix and data home, and time recovery; object store cells flush every 10ms' + type: boolean + default: false + groups: + description: 'JSON list of cell groups, each overriding configs, extra_env, concurrency, rounds, rows, restart, plus idle_seconds and partitions (regions of the loaded table), or mode crash with cycles, kill_after [min, max] and pin, or mode drop with write_seconds; empty = one group from the inputs above' + type: string + default: '' 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 @@ -47,6 +67,11 @@ permissions: env: SMOKE: ${{ inputs.smoke }} GREPTIMEDB_TAG: ${{ inputs.greptimedb_tag }} + GREPTIMEDB_IMAGE_INPUT: ${{ inputs.greptimedb_image }} + CONFIGS: ${{ inputs.configs }} + EXTRA_ENV: ${{ inputs.extra_env }} + RESTART: ${{ inputs.restart }} + OSWAL_GROUPS: ${{ inputs.groups }} REQUESTED_INSTANCE_TYPE: ${{ inputs.instance_type }} CONCURRENCY: ${{ inputs.concurrency }} ROWS: ${{ inputs.rows }} @@ -61,7 +86,7 @@ jobs: timeout-minutes: 15 outputs: instance_type: ${{ steps.runner.outputs.instance_type }} - greptimedb_tag: ${{ steps.plan.outputs.greptimedb_tag }} + greptimedb_ref: ${{ steps.plan.outputs.greptimedb_ref }} runtime_image: ${{ steps.runtime.outputs.runtime_image }} steps: - uses: actions/checkout@v4 @@ -77,8 +102,9 @@ jobs: run: | python3 greptimedb/.github/scripts/oswal-s3-bench.py plan --smoke "$SMOKE" \ --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS" \ + --configs "$CONFIGS" --extra-env "$EXTRA_ENV" --restart "$RESTART" --groups "$OSWAL_GROUPS" \ --timeout-minutes "$BENCHMARK_TIMEOUT_MINUTES" --ttl-hours "$JANITOR_TTL_HOURS" - python3 greptimedb/.github/scripts/oswal-s3-bench.py tag "$GREPTIMEDB_TAG" + python3 greptimedb/.github/scripts/oswal-s3-bench.py image --tag "$GREPTIMEDB_TAG" --image "$GREPTIMEDB_IMAGE_INPUT" - name: Resolve CI runner id: runner run: python3 greptimedb/.github/scripts/ci-runner-config.py --provider AWS --instance-type "$REQUESTED_INSTANCE_TYPE" @@ -132,7 +158,7 @@ jobs: 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 }} + GREPTIMEDB_REF: ${{ needs.validate.outputs.greptimedb_ref }} 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 }} @@ -161,7 +187,8 @@ jobs: 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" + --concurrency "$CONCURRENCY" --rows "$ROWS" --batch-rows "$BATCH_ROWS" --rounds "$ROUNDS" \ + --configs "$CONFIGS" --extra-env "$EXTRA_ENV" --restart "$RESTART" --groups "$OSWAL_GROUPS" 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 diff --git a/tests/perf/test_oswal_s3_bench_scripts.py b/tests/perf/test_oswal_s3_bench_scripts.py index dc81240f1ae..6934ff0db13 100644 --- a/tests/perf/test_oswal_s3_bench_scripts.py +++ b/tests/perf/test_oswal_s3_bench_scripts.py @@ -20,9 +20,12 @@ import json import os import subprocess import sys +import re import tempfile import unittest +import urllib.parse from pathlib import Path +from types import SimpleNamespace SCRIPTS = Path(__file__).parents[2] / ".github/scripts" @@ -35,6 +38,7 @@ def load(name, filename): bench = load("oswal_s3_bench", "oswal-s3-bench.py") +crash = load("oswal_s3_bench_crash", "oswal-s3-bench-crash.py") summary = load("oswal_s3_bench_summary", "oswal-s3-bench-summary.py") ACK_BUCKETS = """# TYPE greptime_logstore_object_store_wal_append_ack_seconds histogram @@ -57,8 +61,9 @@ class PlanTest(unittest.TestCase): 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(plan["corpora"], [{"rows": 2_000_000, "dir": "corpus-2000000", "data_profile": "D", + "generator_args": ["--data-profile", "D"]}]) + self.assertEqual((plan["rows"], plan["batch_rows"]), (2_000_000, 1000)) 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"}, @@ -66,8 +71,8 @@ class PlanTest(unittest.TestCase): 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((plan["rows"], plan["rounds"], plan["corpora"][0]["data_profile"]), (50_000, 1, "custom")) + self.assertEqual(plan["corpora"][0]["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): @@ -76,6 +81,129 @@ class PlanTest(unittest.TestCase): with self.subTest(args=args), self.assertRaises(ValueError): bench.build_plan(*args, smoke=False) + def test_the_baselines_and_the_order_reverse_every_other_round(self): + plan = bench.build_plan("8", "2000000", "1000", "2", smoke=False, configs="R,R0,D,E") + self.assertEqual([c["id"] for c in plan["cells"]], [ + "1-8-R", "1-8-R0", "1-8-D", "1-8-E", "2-8-E", "2-8-D", "2-8-R0", "2-8-R"]) + cells = {c["id"]: c for c in plan["cells"]} + self.assertEqual((cells["1-8-R0"]["env"], cells["1-8-R0"]["wal"]), + ({}, {"PROVIDER": "raft_engine", "SYNC_WRITE": "false", "DIR": "/wal"})) + + def test_extra_env_and_restart_reach_every_cell(self): + for extra in ('{"RUST_BACKTRACE": 1, "RUST_LOG": "info"}', "RUST_BACKTRACE=1\n\nRUST_LOG = info\n"): + plan = bench.build_plan("8", "2000000", "1000", "1", smoke=False, configs="R,D", extra_env=extra, + restart=True) + with self.subTest(extra=extra): + self.assertEqual([c["id"] for c in plan["cells"]], ["1-8-R", "1-8-D"]) + cells = {c["id"]: c for c in plan["cells"]} + self.assertEqual(cells["1-8-D"]["env"], {"RUST_BACKTRACE": "1", "RUST_LOG": "info"}) + self.assertEqual(cells["1-8-D"]["wal"]["FLUSH_INTERVAL"], "10ms") + self.assertNotIn("FLUSH_INTERVAL", cells["1-8-R"]["wal"]) + self.assertTrue(all(c["restart"] for c in plan["cells"])) + + def test_invalid_extra_env_and_configs_are_rejected(self): + for kwargs in ({"extra_env": "GREPTIMEDB_STANDALONE__STORAGE__BUCKET=other"}, {"extra_env": "NO_VALUE"}, + {"extra_env": "GREPTIMEDB_STANDALONE__WAL__FLUSH_INTERVAL=100ms", "restart": True}, + {"extra_env": "GREPTIMEDB_STANDALONE__WAL__SYNC_WRITE=false"}, + {"extra_env": "GREPTIMEDB_STANDALONE__wal__FLUSH_INTERVAL=100ms", "restart": True}, + {"extra_env": "greptimedb_standalone__Storage__bucket=other"}, + {"extra_env": "A=1\nA=2"}, {"extra_env": '{"A": true}'}, {"extra_env": "1A=x"}, + {"configs": "R,X"}, {"configs": "D,D"}): + with self.subTest(kwargs=kwargs), self.assertRaises(ValueError): + bench.build_plan("8", "6000", "1000", "1", smoke=False, **kwargs) + + def test_the_image_is_the_full_reference_or_a_greptimedb_tag(self): + image = "docker.io/greptime/greptimedb-dev:dev-20261013-1791848986-a5a1a9f4" + for valid in (image, "ghcr.io/acme/greptimedb--bench:dev", "ghcr.io/acme/greptimedb__bench:dev", + "localhost:5000/a/b@sha256:" + "c" * 64): + self.assertEqual(bench.resolve_image(valid, "nightly"), valid) + self.assertEqual(bench.resolve_image("", "v1.4.0-nightly-20261012"), "greptime/greptimedb:v1.4.0-nightly-20261012") + for bad in ("greptimedb", "docker.io/greptime/greptimedb", "Docker.io/x/y:1", "x/y:bad tag", + "x/y___z:1", "x/-y:1"): + with self.subTest(bad=bad), self.assertRaises(ValueError): + bench.resolve_image(bad, "nightly") + + def test_groups_inherit_the_inputs_and_keep_their_own_settings(self): + groups = [{"name": "recovery", "configs": "D,E", "rows": 4000000, "rounds": 1, "restart": True}, + {"name": "crashe", "mode": "crash", "configs": "E", "rounds": 1, "cycles": 3, "kill_after": [5, 9]}, + {"name": "pin", "mode": "crash", "configs": "D", "rounds": 1, "cycles": 1, + "kill_after": [540, 540], "pin": True}, + {"name": "idle", "configs": "D", "rounds": 1, "idle_seconds": 180}] + plan = bench.build_plan("8", "2000000", "1000", "2", smoke=False, groups=json.dumps(groups)) + self.assertEqual([c["id"] for c in plan["cells"]], [ + "recovery-1-8-D", "recovery-1-8-E", "crashe-1-8-E", "pin-1-8-D", "idle-1-8-D"]) + self.assertEqual([c["rows"] for c in plan["corpora"]], [4_000_000, 2_000_000]) + cells = {c["id"]: c for c in plan["cells"]} + self.assertEqual(cells["recovery-1-8-D"]["wal"]["FLUSH_INTERVAL"], "10ms") + self.assertEqual((cells["crashe-1-8-E"]["cycles"], cells["crashe-1-8-E"]["kill_after"]), (3, [5, 9])) + self.assertTrue(cells["pin-1-8-D"]["pin"]) + self.assertNotIn("FLUSH_INTERVAL", cells["pin-1-8-D"]["wal"]) + self.assertEqual(cells["idle-1-8-D"]["idle_seconds"], 180) + smoke = bench.build_plan("8", "2000000", "1000", "2", smoke=True, groups=json.dumps(groups)) + pin = next(c for c in smoke["cells"] if c["group"] == "pin") + self.assertEqual((pin["cycles"], pin["kill_after"]), (1, [60, 60])) + self.assertEqual([c["rows"] for c in smoke["corpora"]], [50_000]) + recovery = [{"name": "rec", "configs": "D", "restart": True, "min_objects": 6000}] + self.assertEqual(bench.build_plan("8", "4000000", "1000", "1", smoke=False, + groups=json.dumps(recovery))["cells"][0]["min_objects"], 6000) + self.assertEqual(bench.build_plan("8", "4000000", "1000", "1", smoke=True, + groups=json.dumps(recovery))["cells"][0]["min_objects"], 0) + + def test_the_workflow_passes_groups_through_bash_unchanged(self): + workflow = (SCRIPTS.parent / "workflows" / "oswal-s3-bench.yml").read_text() + self.assertIn("OSWAL_GROUPS: ${{ inputs.groups }}", workflow) + self.assertEqual(workflow.count('--groups "$OSWAL_GROUPS"'), 2) + command = (f'python3 "{SCRIPTS}/oswal-s3-bench.py" plan --smoke false --concurrency 8 --rows 6000 ' + '--batch-rows 1000 --rounds 1 --groups "$OSWAL_GROUPS"') + for groups, expected in (("", "3 cells"), ('[{"name": "a", "configs": "D,E"}]', "2 cells")): + done = subprocess.run(["bash", "-c", command], env=os.environ | {"OSWAL_GROUPS": groups}, + capture_output=True, text=True, timeout=60) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertTrue(done.stdout.startswith(expected), done.stdout) + + def test_a_partitioned_group_and_a_drop_group(self): + groups = [{"name": "multi", "configs": "R,D", "concurrency": "8,32", "partitions": 8}, + {"name": "drop", "mode": "drop", "configs": "D", "rounds": 1, "write_seconds": 45}] + plan = bench.build_plan("8", "2000000", "1000", "1", smoke=False, groups=json.dumps(groups)) + cells = {c["id"]: c for c in plan["cells"]} + self.assertEqual(list(cells), ["multi-1-8-R", "multi-1-8-D", "multi-1-32-R", "multi-1-32-D", "drop-1-8-D"]) + self.assertEqual({c["partitions"] for c in plan["cells"] if c["mode"] == "load"}, {8}) + self.assertEqual((cells["drop-1-8-D"]["write_seconds"], cells["drop-1-8-D"]["mode"]), (45, "drop")) + self.assertNotIn("rows", cells["drop-1-8-D"]) + self.assertEqual([c["rows"] for c in plan["corpora"]], [2_000_000]) + smoke = bench.build_plan("8", "2000000", "1000", "1", smoke=True, groups=json.dumps(groups)) + self.assertEqual(next(c for c in smoke["cells"] if c["mode"] == "drop")["write_seconds"], 10) + self.assertEqual(bench.build_plan("8", "6000", "1000", "1", smoke=False)["cells"][0]["partitions"], 1) + + def test_the_partitioned_table_splits_the_corpus_window_into_equal_ranges(self): + ddl = bench.partitioned_table({"start": "2026-07-10T00:00:00Z", "end": "2026-07-10T01:00:00Z"}, 8) + start = 1783641600000 + bounds = [int(b) for b in re.findall(r"`event_time` < (\d+)", ddl)] + self.assertEqual(bounds, [start + 450_000 * i for i in range(1, 8)]) + self.assertIn(f"`event_time` >= {start + 450_000 * 7}\n)\nENGINE=mito", ddl) + self.assertEqual(ddl.count("`event_time` >="), 7) + self.assertTrue(ddl.startswith("CREATE TABLE `agent_observations` (")) + self.assertIn("PRIMARY KEY (`tenant`, `app`, `environment`)\n)\nPARTITION ON COLUMNS (`event_time`) (", ddl) + self.assertNotIn("{partitions}", ddl) + with self.assertRaises(ValueError): + bench.partitioned_table({"start": "2026-07-10T00:00:00Z", "end": "2026-07-10T00:00:00.004Z"}, 8) + + def test_invalid_groups_are_rejected(self): + for groups in ([], [{"configs": "D"}], [{"name": "Bad"}], [{"name": "a"}, {"name": "a"}], + [{"name": "a", "colour": "red"}], [{"name": "a", "mode": "crash", "configs": "R"}], + [{"name": "a", "mode": "crash", "configs": "D", "restart": True}], + [{"name": "a", "mode": "crash", "configs": "D", "kill_after": [9, 5]}], + [{"name": "a", "mode": "soak"}], [{"name": "a", "idle_seconds": 601}], + [{"name": "a", "configs": "D", "min_objects": 6000}], + [{"name": "a", "configs": "D", "restart": True, "min_objects": -1}], + [{"name": "a", "configs": "D", "partitions": 65}], [{"name": "a", "configs": "D", "partitions": 0}], + [{"name": "a", "mode": "crash", "configs": "D", "partitions": 8}], + [{"name": "a", "mode": "drop", "configs": "R"}], + [{"name": "a", "mode": "drop", "configs": "D", "write_seconds": 0}], + [{"name": "a", "mode": "drop", "configs": "D", "restart": True}]): + with self.subTest(groups=groups), self.assertRaises(ValueError): + bench.build_plan("8", "6000", "1000", "1", smoke=False, groups=json.dumps(groups)) + 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"] @@ -107,6 +235,185 @@ class PlanTest(unittest.TestCase): bench.build_manifest(plan, [{"RepoDigests": []}], "", "", "", "") +def write_lines(path, lines): + path.write_text("".join(f"{line}\n" for line in lines)) + + +def crash_cycle(root, cycle, stripes, acked, requests, exits, found=None): + """One cycle of two writers: attempted stripes, acked batches, requests and exit times per + writer, and what the restart after its kill found, if given.""" + directory = root / f"cycle-{cycle}" + directory.mkdir(parents=True) + for index in range(2): + write_lines(directory / f"attempted-{index}.log", stripes[index]) + write_lines(directory / f"acked-{index}.log", acked[index]) + write_lines(directory / f"requests-{index}.log", requests[index]) + (directory / f"exit-{index}.log").write_text(f"{exits[index]}\n") + if found is not None: + (directory / "found.json").write_text(json.dumps( + {"output": [{"records": {"rows": [[seq, count] for seq, count in found.items()]}}]})) + return directory + + +class CrashOracleTest(unittest.TestCase): + # Writer 0 got batches 0 and 2 acknowledged and lost 4 in flight; writer 1 got 1 and lost 3. + STRIPES = ([0, 2, 4], [1, 3]) + ACKED = ([0, 2], [1]) + REQUESTS = (["0 200 10", "2 200 20", "4 000 105"], ["1 200 15", "3 000 101"]) + + def check(self, root, found, mode="durable", requests=REQUESTS, exits=(106, 102)): + crash_cycle(root, 1, self.STRIPES, self.ACKED, requests, exits) + result = crash.check_kill(root / "cycle-1", 2, 100, 104, 2000) + return result | crash.verify(root, found, 10, mode == "enqueued", 1) + + def test_acked_batches_present_once_and_an_in_flight_extra_pass(self): + with tempfile.TemporaryDirectory() as directory: + result = self.check(Path(directory), {0: 10, 1: 10, 2: 10, 4: 10}) + self.assertEqual((result["kill_errors"], result["verify_errors"]), ([], [])) + self.assertEqual((result["in_flight"], result["writers_at_kill"], result["cycle_extra_batches"], + result["cycle_lost_batches"], result["cycle_present_batches"]), (2, 2, 1, 0, 4)) + + def test_durable_violations_are_reported(self): + cases = {"acknowledged but absent": {0: 10, 1: 10}, "present more than once": {0: 20, 1: 10, 2: 10}, + "present in part": {0: 10, 1: 10, 2: 7}, "never attempted": {0: 10, 1: 10, 2: 10, 9: 10}} + for error, found in cases.items(): + with self.subTest(error=error), tempfile.TemporaryDirectory() as directory: + result = self.check(Path(directory), found) + self.assertTrue(any(error in e for e in result["verify_errors"]), result["verify_errors"]) + + def test_enqueued_loses_acked_batches_without_failing(self): + with tempfile.TemporaryDirectory() as directory: + result = self.check(Path(directory), {0: 10}, mode="enqueued") + self.assertEqual(result["verify_errors"], []) + self.assertEqual((result["cycle_lost_batches"], result["cycle_lost_rows"], result["total_lost_batches"]), + (2, 20, 2)) + + def test_a_later_cycle_reports_only_its_own_losses(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + crash_cycle(root, 1, self.STRIPES, self.ACKED, self.REQUESTS, (106, 102)) + crash_cycle(root, 2, ([10, 12], [11]), ([10], [11]), (["10 200 210", "12 000 305"], ["11 200 215"]), + (306, 302)) + # Cycle 1 lost batches 1 and 2; cycle 2 lost nothing. + result = crash.verify(root, {0: 10, 10: 10, 11: 10}, 10, True, 2) + self.assertEqual((result["cycle_acked_batches"], result["cycle_lost_batches"], + result["total_lost_batches"], result["total_acked_batches"]), (2, 0, 2, 5)) + + def later_cycle(self, root): + """Cycle 1's restart recovered batches 0, 1 and 2 and the in-flight 4; cycle 2 acknowledged + 10 and 11 and had 12 in flight.""" + crash_cycle(root, 1, self.STRIPES, self.ACKED, self.REQUESTS, (106, 102), found={0: 10, 1: 10, 2: 10, 4: 10}) + crash_cycle(root, 2, ([10, 12], [11]), ([10], [11]), (["10 200 210", "12 000 305"], ["11 200 215"]), + (306, 302)) + + def test_a_batch_an_earlier_restart_recovered_must_stay_present(self): + for mode in ("enqueued", "durable"): + with self.subTest(mode=mode), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.later_cycle(root) + # Acknowledged batch 1 and the in-flight 4 vanished after cycle 1's restart found them. + result = crash.verify(root, {0: 10, 2: 10, 10: 10, 11: 10}, 10, mode == "enqueued", 2) + self.assertIn("recovered by an earlier restart but absent: [1, 4]", result["verify_errors"]) + self.assertEqual(result["cycle_lost_batches"], 0) + + def test_only_the_current_cycle_may_lose_acknowledged_batches_in_enqueued_mode(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.later_cycle(root) + found = {0: 10, 1: 10, 2: 10, 4: 10, 10: 10} + result = crash.verify(root, found, 10, True, 2) + self.assertEqual(result["verify_errors"], []) + self.assertEqual((result["cycle_lost_batches"], result["total_lost_batches"]), (1, 1)) + durable = crash.verify(root, found, 10, False, 2) + self.assertEqual(durable["verify_errors"], ["acknowledged but absent: [11]"]) + duplicated = crash.verify(root, found | {10: 20}, 10, True, 2) + self.assertEqual(duplicated["verify_errors"], ["batches present more than once: [10]"]) + + def test_a_connection_lost_before_the_kill_fails(self): + with tempfile.TemporaryDirectory() as directory: + requests = (["0 200 10", "2 200 20", "4 000 50"], ["1 200 15", "3 500 101"]) + result = self.check(Path(directory), {0: 10, 1: 10, 2: 10}, requests=requests, exits=(51, 99)) + self.assertEqual(len(result["kill_errors"]), 3, result["kill_errors"]) + + def test_the_writer_stops_each_stripe_at_its_first_unacknowledged_batch(self): + import http.server + import threading + inserted, lock = [], threading.Lock() + + class Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + body = self.rfile.read(int(self.headers["Content-Length"])).decode() + with lock: + if len(inserted) >= 6: + self.close_connection = True + self.connection.close() + return + inserted.append(body) + self.send_response(200) + self.end_headers() + + def log_message(self, *args): + pass + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + with tempfile.TemporaryDirectory() as directory: + run = Path(directory) + args = SimpleNamespace(dir=str(run / "cycle-1"), url=f"http://127.0.0.1:{server.server_port}/v1/sql", + table="crash", writers=2, first_seq=0, rows=3, seconds=0) + crash.write(args) + found = {} + for body in inserted: + for seq in set(re.findall(r"%28(\d+)%2C", body)): + found[int(seq)] = 3 + result = crash.verify(run, found, 3, False, 1) + self.assertEqual((result["verify_errors"], result["total_acked_batches"]), ([], 6)) + for index in range(2): + attempted = (run / "cycle-1" / f"attempted-{index}.log").read_text().split() + self.assertEqual([int(seq) % 2 for seq in attempted], [index] * len(attempted)) + finally: + server.shutdown() + server.server_close() + + +class WriterDeadlineTest(unittest.TestCase): + def test_writers_stop_at_the_deadline_with_every_batch_acknowledged(self): + import http.server + import threading + status = [200] + + class Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + self.rfile.read(int(self.headers["Content-Length"])) + self.send_response(status[0]) + self.end_headers() + + def log_message(self, *args): + pass + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + with tempfile.TemporaryDirectory() as directory: + run = Path(directory) + args = SimpleNamespace(dir=str(run / "cycle-1"), url=f"http://127.0.0.1:{server.server_port}/", + table="dropped", writers=2, first_seq=0, rows=2, seconds=0.3) + self.assertEqual(crash.write(args), 0) + result = crash.verify(run, {}, 2, True, 1) + self.assertGreater(result["cycle_acked_batches"], 0) + self.assertEqual(result["verify_errors"], []) + for index in range(2): + self.assertTrue((run / "cycle-1" / f"exit-{index}.log").exists()) + # A rejected batch ends a timed writer early, which fails the command. + status[0] = 503 + args.dir, args.seconds = str(run / "cycle-2"), 30 + self.assertEqual(crash.write(args), 1) + finally: + server.shutdown() + server.server_close() + + class SummaryTest(unittest.TestCase): def test_quantile_interpolates_like_prometheus(self): buckets = summary.parse_metrics(ACK_BUCKETS)["buckets"]["append_ack_seconds"] @@ -117,14 +424,14 @@ class SummaryTest(unittest.TestCase): 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=""): + def write_cell(self, root, cell_id, rows_per_second, exit_code=0, metrics="", **extra): 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})) + "wal": bench.WAL_CONFIGS[config], "metrics_scraped": True, "env": {}, + "exit_code": exit_code, "wal_device": "/dev/nvme1n1", "version": "1.4.0", "load_wall_seconds": 61.0} | extra)) 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})) @@ -149,10 +456,172 @@ 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 | 0 | 0 |", output) + self.assertIn("| 1-8-D | — | 80 | 60.0 | 61.0 | 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_ratios_cover_both_controls_and_restart_cells_report_recovery(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + restart = {"restart": True, "restart_scraped": True, "restart_seconds": 3.5, "indexed_before_kill": 117, + "min_objects": 6000, + "restart_query_seconds": 4.25, "restart_count": 6000, + "server_cpu_seconds": 122.0, "host_cpu_busy": 0.5, "host_cpus": 8} + self.write_cell(root, "1-8-R", 100.0) + self.write_cell(root, "1-8-R0", 125.0) + self.write_cell(root, "1-8-D", 50.0, metrics=ACK_BUCKETS, env={"RUST_LOG": "info"}, **restart) + self.write_cell(root, "1-8-E", 75.0) + (root / "cells" / "1-8-D" / "metrics-restart.prom").write_text( + "greptime_logstore_object_store_wal_recovery_seconds_sum 1.5\n" + "greptime_logstore_object_store_wal_recovery_seconds_count 1\n" + "greptime_logstore_object_store_wal_recovered_objects_total 118\n") + rows = [summary.load_cell(path) for path in sorted((root / "cells").iterdir())] + pairs = {p["ratio"]: p["geomean"] for p in summary.ratios(rows)} + self.assertEqual(pairs, {"D/R": 0.5, "D/R0": 0.4, "E/R": 0.75, "E/R0": 0.6}) + cell = next(r for r in rows if r["id"] == "1-8-D") + self.assertEqual(cell["restart"], {"wall_seconds": 3.5, "query_seconds": 4.25, "rows": 6000, + "recovery_seconds": 1.5, "recovered_objects": 118, + "indexed_before_kill": 117, "min_objects": 6000, "replay": None}) + self.assertEqual(cell["server_cpu_cores"], 2.0) + output = summary.render({}, rows, summary.ratios(rows)) + self.assertIn("| 1-8-D | RUST_LOG=info | 120 | 117 (under 6,000) | 3.50 | 4.25 | 1.500 | 118 " + "| 6,000 |", output) + self.assertIn("| 2.00 | 50% of 8 |", output) + self.assertIn("| 8 | E/R0 | 0.600 | 0.600 |", output) + + def test_traffic_idle_crash_and_pinned_cells_are_summarized(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + traffic = ("greptime_logstore_object_store_wal_created_objects_total 600\n" + "greptime_logstore_object_store_wal_object_bytes_sum 3000000000\n" + "greptime_logstore_object_store_wal_object_bytes_count 600\n" + 'opendal_operation_duration_seconds_count{scheme="s3",operation="write"} 650\n' + 'opendal_operation_duration_seconds_count{scheme="s3",operation="read"} 40\n') + # The final scrape follows a 180 s idle window that created 2 objects of 1 MB. + self.write_cell(root, "1-32-E", 90.0, idle_seconds=180, group="", metrics=traffic.replace( + " 600\n", " 602\n", 1).replace("3000000000", "3002000000").replace("count 600", "count 602") + .replace("} 650", "} 652")) + (root / "cells" / "1-32-E" / "metrics-idle.prom").write_text(traffic.replace("} 650", "} 648")) + cell = summary.load_cell(root / "cells" / "1-32-E") + self.assertEqual((cell["objects_per_second"], cell["mean_object_bytes"], cell["sealed_on_size"]), + (10.0, 5e6, False)) + self.assertAlmostEqual(cell["wal_mb_per_second"], 50.0) + self.assertEqual(cell["idle"], {"seconds": 180, "creates": 2, "operations": {"write": 4}}) + path = root / "cells" / "pin-1-8-D" + path.mkdir(parents=True) + (path / "cell.json").write_text(json.dumps({ + "id": "pin-1-8-D", "group": "pin", "mode": "crash", "round": 1, "concurrency": 8, "config": "D", + "wal": bench.WAL_CONFIGS["D"], "exit_code": 0, "metrics_scraped": True})) + (path / "cycles.jsonl").write_text(json.dumps({ + "cycle": 1, "kill_after_seconds": 540.0, "load_seconds": 540.0, "cycle_acked_batches": 4000, + "cycle_present_batches": 4003, "cycle_lost_batches": 0, "cycle_lost_rows": 0, "cycle_extra_batches": 3, + "in_flight": 8, "total_found_batches": 4003, "total_acked_batches": 4000, "total_lost_batches": 0, + "restart_seconds": 4.5, "pin_rows": 1, "passed": True, "metrics_complete": True, + "kill_errors": [], "verify_errors": []}) + "\n") + (path / "metrics-start-1.prom").write_text("greptime_logstore_object_store_wal_created_objects_total 1\n") + (path / "metrics-cycle-1.prom").write_text( + "greptime_logstore_object_store_wal_indexed_objects 5400\n" + "greptime_logstore_object_store_wal_created_objects_total 5401\n" + "greptime_logstore_object_store_wal_object_bytes_sum 10800000000\n" + "greptime_logstore_object_store_wal_object_bytes_count 5400\n" + 'opendal_operation_duration_seconds_count{scheme="s3",operation="write"} 9000\n' + 'opendal_operation_duration_seconds_count{scheme="fs",operation="write"} 70\n') + (path / "metrics-restart-1.prom").write_text( + "greptime_logstore_object_store_wal_recovery_seconds_sum 2.5\n" + "greptime_logstore_object_store_wal_recovery_seconds_count 1\n" + "greptime_logstore_object_store_wal_recovered_objects_total 5399\n") + (path / "objects.tsv").write_text("100 300 0 301\n130 600 0 601\n160 missing\n") + rows = [summary.load_cell(p) for p in sorted((root / "cells").iterdir())] + output = summary.render({}, rows, summary.ratios(rows)) + self.assertIn("| 1-32-E | 50.0 | 10.0 | 4.77 | no | read 40, write 648 | — | 180 | 2 | write 4 |", output) + self.assertIn("| pin-1-8-D | 1 | 540.0 | 4,000 | 4,003 | 0 | 0 | 3 | 8 | 4,003 | 4,000 | 0 | 5,400 " + "| 4.50 | 2.500 | 5,399 | 1 | pass |", output) + self.assertIn("| pin-1-8-D | 1 | 540.0 | 20.0 | 10.0 | 1.91 | no | write 9,000 | — |", output) + self.assertIn("| pin-1-8-D | 300, 600, missing | 1 | 0 | 601 |", output) + + def test_partitioned_load_cells_report_client_latency_entries_and_regions(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + metrics = (ACK_BUCKETS + "greptime_logstore_object_store_wal_object_entries_sum 4000\n" + "greptime_logstore_object_store_wal_object_entries_count 500\n" + "greptime_logstore_object_store_wal_object_bytes_count 500\n") + self.write_cell(root, "1-8-D", 80.0, metrics=metrics, partitions=8) + path = root / "cells" / "1-8-D" + transcript = [{"step": "create_table", "duration_seconds": 9.0}] + transcript += [{"step": "grpc_write_batch", "status": "success", "duration_seconds": d / 1000} + for d in range(1, 101)] + (path / "greptimedb-1" / "load-transcript.jsonl").write_text( + "".join(json.dumps(r) + "\n" for r in transcript)) + (path / "regions.json").write_text(json.dumps({"output": [{"records": {"rows": [ + [f"p{i}", 4398046511104 + i, 750 + i, 1000] for i in range(8)]}}]})) + cell = summary.load_cell(path) + self.assertEqual((cell["client_p50"], cell["client_p99"]), (0.05, 0.099)) + self.assertEqual((cell["entries_per_object"], cell["appends_per_object"]), (8.0, 0.2)) + self.assertEqual(cell["region_rows"], [750 + i for i in range(8)]) + output = summary.render({}, [cell], []) + self.assertIn("| 1-8-D | 50.0 / 99.0 | 8.0 | 0.20 | 8 | 750 / 757 |", output) + + def test_restarts_report_what_each_region_replayed(self): + log = ("INFO mito2::region::opener: Replay WAL for region: 4398046511104(1024, 0), provider: ObjectStore(" + 'ObjectStoreProvider { region_id: 4398046511104(1024, 0), prefix: "wal" }), rows recovered: 0, ' + "replay from entry id: 1048578, last entry id: 1048577, total timeseries replayed: 0, elapsed: 60.581µs\n" + "INFO mito2::region::opener: Replay WAL for region: 4402341478400(1025, 0), provider: ObjectStore(" + 'ObjectStoreProvider { region_id: 4402341478400(1025, 0), prefix: "wal" }), rows recovered: 12625000, ' + "replay from entry id: 4853858306, last entry id: 8164212737, total timeseries replayed: 82, " + "elapsed: 95.150155247s\n" + "INFO mito2::region::opener: Replay WAL for region: 4402341478401(1025, 1), provider: ObjectStore(" + 'ObjectStoreProvider { region_id: 4402341478401(1025, 1), prefix: "wal" }), rows recovered: 5, ' + "replay from entry id: 8164212736, last entry id: 8164212737, total timeseries replayed: 1, " + "elapsed: 12.5ms\n") + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) + (path / "server-restart.log").write_text(log) + metrics = summary.parse_metrics("greptime_logstore_object_store_wal_read_seconds_sum 95.15\n" + "greptime_logstore_object_store_wal_read_seconds_count 3\n") + result = summary.replay(path / "server-restart.log", metrics) + self.assertEqual((result["rows"], result["object_span"], result["read_seconds"], result["reads"]), + (12_625_005, 3158 + 1, 95.15, 3)) + self.assertAlmostEqual(result["seconds"], 95.150155247) + self.assertEqual([(r["region"], r["rows"], r["object_span"]) for r in result["regions"]], + [(4398046511104, 0, 0), (4402341478400, 12_625_000, 3158), (4402341478401, 5, 1)]) + self.assertEqual((result["regions"][1]["first_entry_id"], result["regions"][1]["last_entry_id"]), + (4853858306, 8164212737)) + self.assertAlmostEqual(result["regions"][0]["seconds"], 60.581e-6) + self.assertEqual(summary.replayed(result["regions"]), + "4402341478400: 12,625,000 rows, object sequences 4,629–7,786, 95.15 s; " + "4402341478401: 5 rows, object sequences 7,786–7,786, 0.01 s") + self.assertIsNone(summary.replay(path / "absent.log", metrics)) + raft = log.replace("ObjectStore(", "RaftEngine(") + (path / "server-restart.log").write_text(raft) + self.assertIsNone(summary.replay(path / "server-restart.log", metrics)["object_span"]) + # A region flushed in object 100 that next wrote in object 200 reads one object but spans 101. + gap = log.replace("replay from entry id: 4853858306", f"replay from entry id: {(100 << 20) + 2}") \ + .replace("last entry id: 8164212737", f"last entry id: {(200 << 20) + 1}") + (path / "server-restart.log").write_text(gap) + self.assertEqual(summary.replay(path / "server-restart.log", metrics)["regions"][1]["object_span"], 101) + + def test_a_drop_cell_reports_the_objects_collected_after_its_restart(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + path = root / "cells" / "drop-1-8-D" + (path / "dropped").mkdir(parents=True) + (path / "cell.json").write_text(json.dumps({ + "id": "drop-1-8-D", "group": "drop", "mode": "drop", "round": 1, "concurrency": 8, "config": "D", + "wal": bench.WAL_CONFIGS["D"], "exit_code": 0, "restart_seconds": 2.5, "restart_scraped": True})) + write_lines(path / "dropped" / "acked-0.log", [0, 8, 16]) + write_lines(path / "dropped" / "acked-1.log", [1, 9]) + (path / "metrics-written.prom").write_text("greptime_logstore_object_store_wal_indexed_objects 300\n") + (path / "metrics-restart.prom").write_text( + "greptime_logstore_object_store_wal_recovered_objects_total 280\n" + "greptime_logstore_object_store_wal_deleted_objects_total 279\n") + (path / "collected.tsv").write_text("100 281\n101 120\n104 2\n") + cell = summary.load_cell(path) + self.assertEqual(cell["drop"], {"acked_batches": 5, "indexed_before_drop": 300, "restart_seconds": 2.5, + "recovered_objects": 280, "indexed_at_end": 2, + "deleted_after_restart": 279, "seconds_to_collect": 4}) + output = summary.render({}, [cell], []) + self.assertIn("| drop-1-8-D | 5 | 300 | 2.50 | 280 | 2 | 279 | 4 | collected |", output) + def test_untouched_counters_are_zero_only_for_an_active_object_store_wal(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) @@ -199,25 +668,54 @@ 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 [[ "$arg" != -d ]] || { echo $(($(cat "$FAKE_LOG.runs" 2>/dev/null || echo 0) + 1)) > "$FAKE_LOG.runs"; rm -f "${FAKE_KILL_FLAG:-/nonexistent}"; 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}" \ + echo "{\"count_before\": 0, \"count_after\": $FAKE_ROWS, \"load_seconds\": 2, \"rows_per_second\": 3000, \"table\": \"agent_observations\"}" \ > "$results/greptimedb-1/load-result.json" ;; + kill) touch "${FAKE_KILL_FLAG:-/dev/null}" ;; inspect) echo true ;; - exec) echo 'version: 1.4.0-test' ;; - logs) echo 'server log' ;; + exec) [[ "$*" == *cpu.stat* ]] && echo 'usage_usec 5000000' || echo 'version: 1.4.0-test' ;; + logs) echo 'server log' + # The rows a server replayed are its start number, so each log can be told apart. + [[ -z "${FAKE_REPLAY:-}" ]] || echo "INFO Replay WAL for region: 4398046511104(1024, 0), provider: ObjectStore(x), rows recovered: $(cat "$FAKE_LOG.runs"), replay from entry id: 1048577, last entry id: 2097153, total timeseries replayed: 1, elapsed: 1.5s" ;; esac """ FAKE_CURL = r"""#!/usr/bin/env bash printf '%s\n' "curl $*" >> "$FAKE_LOG" +[[ "$*" != *"--max-time 10"* || "${FAKE_SAMPLE_RC:-0}" == 0 ]] || exit "$FAKE_SAMPLE_RC" for arg in "$@"; do + [[ "$arg" != *"GROUP BY seq"* ]] || { python3 -c 'import collections, json, sys +counts = collections.Counter(int(l) for l in open(sys.argv[1]) if l.strip()) +print(json.dumps({"output": [{"records": {"rows": [[k, v * int(sys.argv[2])] for k, v in counts.items()]}}]}))' \ + "$FAKE_FOUND" "$FAKE_BATCH_ROWS"; exit 0; } +done +for arg in "$@"; do + [[ "$arg" != *region_statistics* ]] || { python3 -c 'import json, sys +print(json.dumps({"output": [{"records": {"rows": [[f"p{i}", i, int(n), 1] for i, n in enumerate(sys.argv[1].split(","))]}}]}))' \ + "$FAKE_REGION_ROWS"; exit 0; } +done +for arg in "$@"; do + [[ "$arg" != */v1/sql* ]] || { echo "{\"output\": [{\"records\": {\"rows\": [[${FAKE_COUNT:-6000}]]}}]}"; 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; + printf '%s\n' 'greptime_logstore_object_store_wal_poisoned_total 1' 'other_total 2' + [[ -z "${FAKE_FULL_METRICS:-}" ]] || printf '%s\n' \ + "greptime_logstore_object_store_wal_indexed_objects $FAKE_FULL_METRICS" \ + 'greptime_logstore_object_store_wal_recovery_seconds_count 1' \ + "greptime_logstore_object_store_wal_recovered_objects_total ${FAKE_RECOVERED:-0}" + # Every server counts the objects it collects from its start: with FAKE_COLLECTING one more on + # every scrape; with FAKE_FIRST_COLLECTS_AFTER the first server collects one object once it has + # been scraped that many times. + run=$(cat "$FAKE_LOG.runs") n=$(($(cat "$FAKE_LOG.scrapes.$run" 2>/dev/null || echo 0) + 1)) + echo "$n" > "$FAKE_LOG.scrapes.$run" + [[ -z "${FAKE_COLLECTING:-}" ]] || echo "greptime_logstore_object_store_wal_deleted_objects_total $((n - 1))" + [[ -z "${FAKE_FIRST_COLLECTS_AFTER:-}" || "$run" != 1 || "$n" -le "$FAKE_FIRST_COLLECTS_AFTER" ]] \ + || echo 'greptime_logstore_object_store_wal_deleted_objects_total 1' + exit 0; } done """ @@ -225,19 +723,20 @@ done class CellLifecycleTest(unittest.TestCase): # the run script against fake docker and curl - def run_cells(self, root, load_rc, metrics_rc=0): + def run_cells(self, root, load_rc, metrics_rc=0, restart=False, count=6000, groups=""): 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"] + plan = bench.build_plan("8", "6000", "1000", "1", smoke=False, configs="D,E", restart=restart, + extra_env="RUST_LOG=info", groups=groups) (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"), + "FAKE_LOAD_RC": str(load_rc), "FAKE_METRICS_RC": str(metrics_rc), "FAKE_ROWS": "6000", "FAKE_COUNT": str(count), + "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", @@ -302,6 +801,9 @@ class CellLifecycleTest(unittest.TestCase): # the run script against fake docke 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) + self.assertIn("RUST_LOG=info", config) + self.assertIn("-e RUST_LOG=info ", log) + self.assertNotIn("docker kill", log) recorded = json.loads((cell / "cell.json").read_text()) self.assertEqual((recorded["exit_code"], recorded["version"], recorded["metrics_scraped"]), (0, "1.4.0-test", True)) @@ -313,6 +815,260 @@ class CellLifecycleTest(unittest.TestCase): # the run script against fake docke self.assertNotEqual(leaked.returncode, 0) self.assertFalse((root / "results").exists()) + def test_a_restart_cell_kills_restarts_and_counts_the_rows(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, log, _ = self.run_cells(root, load_rc=0, restart=True) + self.assertEqual(cells.returncode, 0, cells.stdout + cells.stderr) + self.assertEqual(log.count("docker kill --signal KILL fakecid"), 2) + self.assertEqual(log.count("docker stop -t 120 fakecid"), 2) + self.assertIn("--name o11ybench-oswal-1-1-1-8-D-restart", log) + self.assertIn("sql=SELECT count(*) FROM agent_observations", log) + cell = root / "results" / "cells" / "1-8-D" + recorded = json.loads((cell / "cell.json").read_text()) + self.assertEqual((recorded["exit_code"], recorded["restart_count"], recorded["restart_scraped"]), (0, 6000, True)) + self.assertIsInstance(recorded["restart_seconds"], float) + config = (cell / "config.env").read_text().splitlines() + self.assertIn("GREPTIMEDB_STANDALONE__STORAGE__DATA_HOME=/oswal_home", config) + self.assertIn("GREPTIMEDB_STANDALONE__WAL__FLUSH_INTERVAL=10ms", config) + self.assertIn(f"type=bind,src={root}/wal/1-8-D-home,dst=/oswal_home", log) + for name in ("metrics.prom", "metrics-restart.prom", "server.log", "server-restart.log"): + self.assertTrue((cell / name).exists(), name) + + def test_an_idle_window_scrapes_before_and_after_it(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + groups = json.dumps([{"name": "idle", "configs": "D", "idle_seconds": 1}]) + cells, _, log, _ = self.run_cells(root, load_rc=0, groups=groups) + self.assertEqual(cells.returncode, 0, cells.stdout + cells.stderr) + cell = root / "results" / "cells" / "idle-1-8-D" + self.assertTrue((cell / "metrics-idle.prom").exists()) + self.assertTrue((cell / "metrics.prom").exists()) + self.assertEqual(log.count("/metrics"), 2) + self.assertIn(f"type=bind,src={root}/corpus/corpus-6000,dst=/work/corpus,readonly", log) + + def run_crash(self, root, metrics_rc=0, pin=False, sample_rc=0, full_metrics="5", groups=None, env=None, + status=200): + """A crash cell, or another group that writes through the crash writer, against fake docker + and a stub SQL endpoint that the fake kill takes down.""" + import http.server + import threading + flag, found = root / "killed", root / "found.log" + found.write_text("") + lock = threading.Lock() + + class Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + body = urllib.parse.unquote_plus(self.rfile.read(int(self.headers["Content-Length"])).decode()) + if flag.exists(): + self.close_connection = True + self.connection.close() + return + seq = re.search(r"VALUES [(](\d+),", body).group(1) + with lock, open(found, "a") as out: + out.write(seq + "\n") + self.send_response(status) + self.end_headers() + + def log_message(self, *args): + pass + + class Server(http.server.ThreadingHTTPServer): + request_queue_size = 64 # all eight writers connect at once + + server = Server(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + groups = groups or json.dumps([{"name": "crash", "configs": "D", "mode": "crash", "cycles": 2, + "kill_after": [1, 1], "pin": pin}]) + old = os.environ.copy() + os.environ.update({"OSWAL_SQL_URL": f"http://127.0.0.1:{server.server_port}/v1/sql", + "FAKE_KILL_FLAG": str(flag), "FAKE_FOUND": str(found), "FAKE_BATCH_ROWS": "1000", + "OSWAL_SAMPLE_SECONDS": "0.2", "FAKE_SAMPLE_RC": str(sample_rc), + "FAKE_FULL_METRICS": full_metrics} | (env or {})) + try: + return self.run_cells(root, load_rc=0, metrics_rc=metrics_rc, groups=groups, + count=(env or {}).get("FAKE_COUNT", 6000)) + finally: + os.environ.clear() + os.environ.update(old) + server.shutdown() + server.server_close() + + def test_a_crash_cell_kills_restarts_and_checks_every_cycle(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, log, _ = self.run_crash(root, env={"FAKE_REPLAY": "1"}) + self.assertEqual(cells.returncode, 0, cells.stdout + cells.stderr) + cell = root / "results" / "cells" / "crash-1-8-D" + # Server 1 is loaded and killed; servers 2 and 3 are the restarts of cycles 1 and 2. + replays = [c["replay"]["rows"] for c in summary.load_cell(cell)["cycles"]] + self.assertEqual(replays, [2, 3]) + cycles = [json.loads(line) for line in (cell / "cycles.jsonl").read_text().splitlines()] + self.assertEqual([c["cycle"] for c in cycles], [1, 2]) + for c in cycles: + self.assertTrue(c["passed"] and c["metrics_complete"], c) + self.assertEqual((c["cycle_lost_batches"], c["writers_at_kill"]), (0, 8)) + self.assertGreater(c["cycle_acked_batches"], 0) + self.assertEqual(log.count("docker kill --signal KILL fakecid"), 2) + self.assertEqual(log.count("docker wait fakecid"), 2) + self.assertIn("GREPTIMEDB_STANDALONE__STORAGE__DATA_HOME=/oswal_home", (cell / "config.env").read_text()) + for name in ("metrics-start-1.prom", "metrics-cycle-2.prom", "metrics-restart-2.prom", "server.log", + "server-restart-1.log", "server-restart-2.log"): + self.assertTrue((cell / name).exists(), name) + + def test_failed_pinned_samples_are_missing_not_zero(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, log, _ = self.run_crash(root, pin=True, sample_rc=28, env={"FAKE_COUNT": "1"}) + self.assertEqual(cells.returncode, 0, cells.stdout + cells.stderr) + cell = root / "results" / "cells" / "crash-1-8-D" + samples = (cell / "objects.tsv").read_text().splitlines() + self.assertTrue(samples and all(line.split()[1:] == ["missing"] for line in samples), samples) + self.assertIn("sql=INSERT INTO pin VALUES (1, 0)", log) + self.assertIn("WITH ('auto_flush_interval' = '1d')", urllib.parse.unquote_plus(log)) + self.assertEqual(len((cell / "cycles.jsonl").read_text().splitlines()), 2) + + def test_a_pinned_cell_fails_when_objects_the_pin_held_are_collected(self): + # Each server collects two objects before its kill and no periodic sample is taken. With one + # object below the pin the first server already exceeds it; with two, only both together do. + for below_pin, failed_cycle in (("1", 1), ("2", 2)): + with self.subTest(below_pin=below_pin), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, _, _ = self.run_crash(root, pin=True, full_metrics=below_pin, env={ + "FAKE_COUNT": "1", "FAKE_COLLECTING": "1", "OSWAL_SAMPLE_SECONDS": "30"}) + cell = root / "results" / "cells" / "crash-1-8-D" + self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], 11, + cells.stdout + cells.stderr) + self.assertFalse((cell / "objects.tsv").exists()) + self.assertEqual(len((cell / "cycles.jsonl").read_text().splitlines()), failed_cycle) + self.assertIn(f"cycle {failed_cycle} collected WAL objects the pin held", cells.stdout + cells.stderr) + + def test_a_pinned_cell_may_collect_the_objects_below_its_pin_late(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + # The start object, the one object indexed before the pin, goes only after the first cycle's + # baseline: the pin scrape and that baseline are the first server's first two scrapes. + cells, _, _, _ = self.run_crash(root, pin=True, full_metrics="1", env={ + "FAKE_COUNT": "1", "FAKE_FIRST_COLLECTS_AFTER": "2", "OSWAL_SAMPLE_SECONDS": "30"}) + cell = root / "results" / "cells" / "crash-1-8-D" + self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], 0, cells.stdout + cells.stderr) + self.assertNotIn("deleted_objects_total", (cell / "metrics-start-1.prom").read_text()) + self.assertIn("deleted_objects_total 1", (cell / "metrics-cycle-1.prom").read_text()) + self.assertEqual(len((cell / "cycles.jsonl").read_text().splitlines()), 2) + + def test_a_durable_pinned_cycle_fails_unless_its_pin_row_is_there_once(self): + for count in ("0", "2"): + with self.subTest(count=count), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.run_crash(root, pin=True, env={"FAKE_COUNT": count}) + cell = root / "results" / "cells" / "crash-1-8-D" + cycle = json.loads((cell / "cycles.jsonl").read_text().splitlines()[0]) + self.assertEqual((cycle["passed"], cycle["pin_rows"]), (False, int(count))) + self.assertIn(f"the pin table holds {count} rows, not 1", cycle["verify_errors"]) + self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], 6) + + def test_a_scrape_without_the_wal_samples_is_incomplete(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, _, _ = self.run_crash(root, full_metrics="") + self.assertNotEqual(cells.returncode, 0) + cell = root / "results" / "cells" / "crash-1-8-D" + cycle = json.loads((cell / "cycles.jsonl").read_text().splitlines()[0]) + self.assertEqual((cycle["passed"], cycle["metrics_complete"]), (True, False)) + self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], 4) + + def test_a_restart_cell_must_leave_its_minimum_objects_indexed(self): + for minimum, exit_code in ((5, 0), (6, 7)): + with self.subTest(minimum=minimum), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + groups = json.dumps([{"name": "rec", "configs": "D", "restart": True, "min_objects": minimum}]) + old = os.environ.copy() + os.environ["FAKE_FULL_METRICS"] = "5" + try: + cells, _, _, _ = self.run_cells(root, load_rc=0, groups=groups) + finally: + os.environ.clear() + os.environ.update(old) + recorded = json.loads((root / "results" / "cells" / "rec-1-8-D" / "cell.json").read_text()) + self.assertEqual((recorded["exit_code"], recorded["indexed_before_kill"], recorded["restart_count"]), + (exit_code, 5, 6000)) + + def test_a_partitioned_cell_creates_its_table_and_checks_every_region_took_rows(self): + for region_rows, exit_code in (("750,750,750,750,750,750,750,750", 0), ("1000,1000,1000,0,1000,1000,1000,0", 8)): + with self.subTest(region_rows=region_rows), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "corpus" / "corpus-6000").mkdir(parents=True) + (root / "corpus" / "corpus-6000" / "summary.json").write_text(json.dumps( + {"start": "2026-07-10T00:00:00Z", "end": "2026-07-10T01:00:00Z"})) + groups = json.dumps([{"name": "multi", "configs": "D", "partitions": 8}]) + old = os.environ.copy() + os.environ["FAKE_REGION_ROWS"] = region_rows + try: + cells, _, log, _ = self.run_cells(root, load_rc=0, groups=groups) + finally: + os.environ.clear() + os.environ.update(old) + cell = root / "results" / "cells" / "multi-1-8-D" + self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], exit_code, + cells.stdout + cells.stderr) + self.assertIn("PARTITION ON COLUMNS (`event_time`)", urllib.parse.unquote_plus(log)) + self.assertIn(f"type=bind,src={SCRIPTS}/oswal-s3-bench-load.sh,dst=/work/oswal-s3-bench-load.sh,readonly", + log) + self.assertIn(" /work/oswal-s3-bench-load.sh", log) + self.assertNotIn("greptimedb.sh load", log) + self.assertEqual(len(json.loads((cell / "regions.json").read_text())["output"][0]["records"]["rows"]), 8) + + def test_a_drop_cell_restarts_and_waits_for_the_dropped_objects_to_go(self): + for indexed, recovered, exit_code in (("2", "0", 0), ("50", "200", 0), ("50", "60", 9)): + with self.subTest(indexed=indexed, recovered=recovered), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + groups = json.dumps([{"name": "drop", "mode": "drop", "configs": "D", "write_seconds": 1}]) + cells, _, log, _ = self.run_crash(root, full_metrics=indexed, groups=groups, + env={"FAKE_RECOVERED": recovered, "OSWAL_COLLECT_SECONDS": "1"}) + cell = root / "results" / "cells" / "drop-1-8-D" + recorded = json.loads((cell / "cell.json").read_text()) + self.assertEqual(recorded["exit_code"], exit_code, cells.stdout + cells.stderr) + self.assertIsInstance(recorded["restart_seconds"], float) + sql = urllib.parse.unquote_plus(log) + for statement in ("CREATE TABLE dropped", "CREATE TABLE kept", "DROP TABLE dropped", + "ADMIN FLUSH_TABLE('kept')", "INSERT INTO kept VALUES (-1, "): + self.assertIn(statement, sql) + self.assertLess(sql.index("DROP TABLE dropped"), log.index("docker kill --signal KILL fakecid")) + self.assertEqual(log.count("docker kill --signal KILL fakecid"), 1) + self.assertTrue(list((cell / "dropped").glob("acked-*.log"))) + self.assertTrue(list((cell / "kept").glob("acked-*.log"))) + for name in ("metrics-written.prom", "metrics-restart.prom", "server.log", + "server-restart.log", "collected.tsv"): + self.assertTrue((cell / name).exists(), name) + + def test_a_drop_cell_whose_writers_fail_stops_before_the_drop(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + groups = json.dumps([{"name": "drop", "mode": "drop", "configs": "D", "write_seconds": 1}]) + cells, _, log, _ = self.run_crash(root, full_metrics="2", groups=groups, status=503) + recorded = json.loads((root / "results" / "cells" / "drop-1-8-D" / "cell.json").read_text()) + self.assertEqual(recorded["exit_code"], 10, cells.stdout + cells.stderr) + self.assertNotIn("DROP TABLE", urllib.parse.unquote_plus(log)) + self.assertNotIn("docker kill", log) + + def test_a_crash_cycle_without_its_scrapes_fails_the_cell(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, _, _ = self.run_crash(root, metrics_rc=22) + self.assertNotEqual(cells.returncode, 0) + cell = root / "results" / "cells" / "crash-1-8-D" + cycle = json.loads((cell / "cycles.jsonl").read_text().splitlines()[0]) + self.assertEqual((cycle["passed"], cycle["metrics_complete"]), (True, False)) + self.assertEqual(json.loads((cell / "cell.json").read_text())["exit_code"], 4) + + def test_a_restart_that_loses_rows_fails_the_cell(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cells, _, _, _ = self.run_cells(root, load_rc=0, restart=True, count=5999) + self.assertNotEqual(cells.returncode, 0) + recorded = json.loads((root / "results" / "cells" / "1-8-D" / "cell.json").read_text()) + self.assertEqual((recorded["exit_code"], recorded["restart_count"]), (5, 5999)) + if __name__ == "__main__": unittest.main()