mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-24 12:52:29 +00:00
jeremyhi
a436213bf3
* ci: compare several images in one object store WAL S3 benchmark run
A new `images` input takes a JSON list of 2..4 {label, image}. Every
cell then runs once per image, back to back, and the image order
reverses with the configuration order every other round. The cell id
gains the label. The single-image inputs keep their meaning and cell ids.
The validate job pulls every image before an instance is allocated. It
pins each image to its digest and records the version and commit that
the image reports. Each cell starts the image of its label. An optional
`warmup` runs the first load cell once per image as round 0, and no
statistic uses it.
The plan step estimates the benchmark job's minutes on c6id.2xlarge
from the cell times of earlier runs. It refuses a plan that does not
fit the job timeout.
The summary keeps the Raft Engine ratios within each image. With
several images it also reports, per group, concurrency and config:
- every round's rows/s and client p50/p99;
- each image's ratio to the first, with the geometric mean, minimum
and maximum over the rounds.
Every cell records the host load averages at its start and end.
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* ci: scale the S3 benchmark time estimate with the batch size
The per-cell rates were measured with 1000-row batches. A smaller batch
keeps about the request rate, not the row rate, so the estimate now
shrinks rows/s with the batch. A one-row durable plan is refused instead
of passing a timeout it cannot fit.
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* ci: pin the S3 benchmark runtime image without pulling it in validate
An anonymous pull from ECR Public on a hosted runner can exceed the
registry's data limit, and it failed the validate job of several
dispatches with "toomanyrequests: Data limit exceeded". Validate now
reads only the registry manifest to pin the runtime image's digest. A
reference that names a digest must be served under that digest.
The check that the image holds the loader and the generator moves to
the benchmark host. It runs right after the runtime pull and before any
GreptimeDB image is pulled or any cell starts.
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* test: run the S3 benchmark runtime content check against both files
The fake docker now runs the pull step's own check in a directory that stands in for the runtime image. It covers both files present, the loader missing and the generator missing, so dropping either file check fails the test.
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* ci: pin a digest-named S3 benchmark runtime image without the registry
ECR Public's anonymous data limit now refuses hosted runners even the
manifest read, and validate failed in "Pin the runtime image digest".
A runtime reference that names its sha256 digest, like the default one,
is already pinned, so validate takes that digest with no network call.
The format checks stay. Only a tag-only reference is resolved through
the registry. The content check on the benchmark host is unchanged.
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* ci: check the full S3 benchmark runtime reference before pinning it offline
A named digest no longer reaches the registry, so nothing parsed the rest of the reference. Validate now checks the whole reference against Docker's reference grammar first. It refuses an empty repository, a doubled '@' or a bad tag before any pin is emitted or any instance is allocated.
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
---------
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
522 lines
32 KiB
Python
522 lines
32 KiB
Python
#!/usr/bin/env python3
|
||
# Copyright 2023 Greptime Team
|
||
#
|
||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||
# you may not use this file except in compliance with the License.
|
||
# You may obtain a copy of the License at
|
||
#
|
||
# http://www.apache.org/licenses/LICENSE-2.0
|
||
#
|
||
# Unless required by applicable law or agreed to in writing, software
|
||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
# See the License for the specific language governing permissions and
|
||
# limitations under the License.
|
||
|
||
"""Summarize the object store WAL S3 benchmark as Markdown and summary.json."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import math
|
||
import re
|
||
from pathlib import Path
|
||
|
||
PREFIX = "greptime_logstore_object_store_wal_"
|
||
COUNTERS = ("created_objects_total", "stalled_appends_total", "poisoned_total",
|
||
"create_failures_total", "create_conflicts_total")
|
||
# 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, 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))
|
||
le = re.search(r'le="([^"]+)"', labels)
|
||
if name.endswith("_bucket") and le:
|
||
buckets.setdefault(name[:-len("_bucket")], []).append((float(le.group(1)), value))
|
||
else:
|
||
samples[name] = value
|
||
return {"samples": samples, "buckets": {name: sorted(items) for name, items in buckets.items()},
|
||
"operations": operations}
|
||
|
||
|
||
def quantile(q: float, buckets: list) -> float | None:
|
||
"""Prometheus histogram_quantile over cumulative (upper bound, count) buckets."""
|
||
if not buckets or buckets[-1][1] == 0:
|
||
return None
|
||
rank = q * buckets[-1][1]
|
||
lower, below = 0.0, 0.0
|
||
for upper, count in buckets:
|
||
if count >= rank:
|
||
if math.isinf(upper):
|
||
return lower
|
||
return lower + (upper - lower) * (rank - below) / (count - below) if count > below else upper
|
||
lower, below = upper, count
|
||
return lower
|
||
|
||
|
||
def geomean(values: list) -> float | None:
|
||
return math.exp(sum(map(math.log, values)) / len(values)) if values and all(v > 0 for v in values) else None
|
||
|
||
|
||
def 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 {}
|
||
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"], "group": cell.get("group", ""), "mode": cell.get("mode", "load"),
|
||
"image": cell.get("image", ""), "warmup": cell.get("warmup", False),
|
||
"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), "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,
|
||
"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"),
|
||
"host_load_start": cell.get("host_load_start"), "host_load_end": cell.get("host_load_end"),
|
||
}
|
||
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"
|
||
and metrics["samples"].get("created_objects_total"))
|
||
for counter in COUNTERS:
|
||
row[counter] = metrics["samples"].get(counter, 0 if active else None)
|
||
created = row["created_objects_total"]
|
||
row["creates_per_minute"] = created * 60 / load_seconds if created is not None and load_seconds else None
|
||
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:
|
||
"""Each object store WAL cell over each Raft Engine control of the same image per round, and their
|
||
geometric mean over the rounds, per concurrency; warm-up cells are left out."""
|
||
rows = [r for r in rows if r["mode"] == "load" and not r["warmup"]]
|
||
speed = {(r["group"], r["image"], r["round"], r["concurrency"], r["config"]): r["rows_per_second"]
|
||
for r in rows}
|
||
result = []
|
||
for group, image in dict.fromkeys((r["group"], r["image"]) for r in rows):
|
||
members = [r for r in rows if (r["group"], r["image"]) == (group, image)]
|
||
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, image, round_number, concurrency, baseline))
|
||
measured = speed.get((group, image, 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, "image": image, "concurrency": concurrency,
|
||
"ratio": f"{config}/{baseline}", "per_round": per_round,
|
||
"geomean": geomean(complete) if len(complete) == len(per_round) else None})
|
||
return result
|
||
|
||
|
||
def image_ratios(rows: list, images: list) -> list:
|
||
"""Every load cell of every image per round, and each image's rows/s over the first image's in the
|
||
same round, with their geometric mean, minimum and maximum over the rounds; warm-up cells are left out."""
|
||
rows = [r for r in rows if r["mode"] == "load" and not r["warmup"]]
|
||
cells = {(r["group"], r["image"], r["round"], r["concurrency"], r["config"]): r for r in rows}
|
||
result = []
|
||
for group, concurrency, config in dict.fromkeys((r["group"], r["concurrency"], r["config"]) for r in rows):
|
||
rounds = sorted({r["round"] for r in rows if (r["group"], r["concurrency"], r["config"]) == (group, concurrency, config)})
|
||
for image in images:
|
||
measured = [cells.get((group, image, n, concurrency, config)) or {} for n in rounds]
|
||
reference = [cells.get((group, images[0], n, concurrency, config)) or {} for n in rounds]
|
||
per_round = [ratio(m.get("rows_per_second"), r.get("rows_per_second")) for m, r in zip(measured, reference)]
|
||
complete = image != images[0] and None not in per_round
|
||
result.append({
|
||
"group": group, "concurrency": concurrency, "config": config, "image": image,
|
||
"rounds": [{"round": n, "id": m.get("id"), "rows_per_second": m.get("rows_per_second"),
|
||
"client_p50": m.get("client_p50"), "client_p99": m.get("client_p99"),
|
||
"host_load_start": m.get("host_load_start"), "host_load_end": m.get("host_load_end")}
|
||
for n, m in zip(rounds, measured)],
|
||
"ratio_to": None if image == images[0] else images[0],
|
||
"per_round": None if image == images[0] else per_round,
|
||
"geomean": geomean(per_round) if complete else None,
|
||
"min": min(per_round) if complete else None, "max": max(per_round) if complete 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}"
|
||
|
||
|
||
def load_average(value) -> str:
|
||
"""The 1-minute load average of a recorded [1, 5, 15] minute triple."""
|
||
return "—" if not value else f"{value[0]:.2f}"
|
||
|
||
|
||
def render(run: dict, rows: list, pairs: list, compared: list = ()) -> str:
|
||
lines = [f"## Object store WAL against S3 ({'smoke, MinIO' if run.get('smoke') else run.get('instance_type')})", ""]
|
||
images = run.get("greptimedb_images") or []
|
||
if len(images) > 1:
|
||
lines += [f"{fmt(run.get('rows'))} rows in {run.get('batch_rows')}-row batches; bucket region {run.get('region')}; "
|
||
f"each image is compared with the first, `{images[0]['label']}`.", "",
|
||
"| Image | Reference | Version | Commit |", "| --- | --- | --- | --- |"]
|
||
lines += [f"| {i['label']} | `{i['image']}` | {i['version']} | `{i['commit']}` |" for i in images] + [""]
|
||
else:
|
||
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 | 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 | Load average at start / end |",
|
||
"| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: "
|
||
"| ---: | ---: | --- | ---: |"]
|
||
def label(r):
|
||
cell = f"{r['id']} (warm-up)" if r["warmup"] else r["id"]
|
||
return f"{cell} (not run)" if r["missing"] else cell if r["exit_code"] == 0 \
|
||
else f"{cell} (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}", 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:]), 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 '—'}`",
|
||
f"{load_average(r['host_load_start'])} / {load_average(r['host_load_end'])} |"]))
|
||
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 | Image | Concurrency | Ratio | Per round | Geometric mean |",
|
||
"| --- | --- | ---: | --- | --- | ---: |"]
|
||
for p in pairs:
|
||
lines.append(f"| {p['group'] or '—'} | {p['image'] or '—'} | {p['concurrency']} | {p['ratio']} "
|
||
f"| {', '.join(fmt(v, 3) for v in p['per_round'])} | {fmt(p['geomean'], 3)} |")
|
||
if compared:
|
||
lines += ["", f"| Group | Concurrency | Config | Image | Rows/s per round | Client p50 / p99 ms per round "
|
||
f"| Ratio to {compared[0]['image']} per round | Geometric mean | Min | Max |",
|
||
"| --- | ---: | --- | --- | --- | --- | --- | ---: | ---: | ---: |"]
|
||
for c in compared:
|
||
lines.append(" | ".join([
|
||
f"| {c['group'] or '—'}", str(c["concurrency"]), c["config"], c["image"],
|
||
"; ".join(fmt(r["rows_per_second"]) for r in c["rounds"]),
|
||
"; ".join(pair(r["client_p50"], r["client_p99"], 1, 1000) for r in c["rounds"]),
|
||
"—" if c["per_round"] is None else ", ".join(fmt(v, 3) for v in c["per_round"]),
|
||
fmt(c["geomean"], 3), fmt(c["min"], 3), fmt(c["max"], 3)]) + " |")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--root", type=Path, required=True)
|
||
args = parser.parse_args()
|
||
args.root.mkdir(parents=True, exist_ok=True)
|
||
run_path = args.root / "run.json"
|
||
run = json.loads(run_path.read_text()) if run_path.exists() else {}
|
||
plan_path = args.root / "plan.json"
|
||
plan = json.loads(plan_path.read_text()) if plan_path.exists() else {}
|
||
planned = plan.get("cells", [])
|
||
# The plan, not what happened to run, defines the expected cells and rounds.
|
||
rows = [load_cell(args.root / "cells" / spec["id"], spec) for spec in planned]
|
||
rows += [load_cell(path) for path in sorted((args.root / "cells").glob("*"))
|
||
if path.name not in {spec["id"] for spec in planned} and (path / "cell.json").exists()]
|
||
pairs = ratios(rows)
|
||
images = plan.get("images", [""])
|
||
compared = image_ratios(rows, images) if len(images) > 1 else []
|
||
missing = [r["id"] for r in rows if r["missing"]]
|
||
(args.root / "summary.json").write_text(json.dumps(
|
||
{"run": run, "planned_cells": len(planned), "missing_cells": missing, "cells": rows, "ratios": pairs,
|
||
"image_ratios": compared}, indent=2) + "\n")
|
||
print(render(run, rows, pairs, compared) if rows else "No benchmark cell completed.\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|