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()