From 97e3bfd547b9194ca10234eda466710ea4239e92 Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Thu, 15 Oct 2026 09:52:28 +0000 Subject: [PATCH] test: add manual S3 metric snapshot benchmark (#9573) * test: add repeatable metric snapshot S3 benchmark Signed-off-by: jeremyhi * test: use only existing CI S3 permissions in the metric snapshot benchmark The CI S3 user is not allowed GetBucketLocation or ListBucketMultipartUploads, so the bucket probe and the prefix cleanup failed with AccessDenied. Use only list, put, get and delete under the run prefix, as the object store WAL benchmark does, and probe S3 before allocating the EC2 runner. Signed-off-by: jeremyhi * test: scope metric snapshot inventory to the current round The full round's S3 inventory listed the whole run prefix, so it also counted the smoke snapshot. List only the round's snapshot prefix while cleanup still covers the run, and assert that the teardown's S3 cleanup step runs always() with the validated run prefix. Signed-off-by: jeremyhi * docs: record the AWS validation run of the metric snapshot benchmark Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi (cherry picked from commit 7df99f946e8c35f565705905734076121c89131d) --- .github/scripts/metric-snapshot-fixture.py | 294 +++++++++++++ .github/scripts/metric-snapshot-s3-bench.py | 392 ++++++++++++++++++ .github/scripts/metric-snapshot-s3-ops.py | 217 ++++++++++ .../tests/test_metric_snapshot_fixture.py | 101 +++++ .../tests/test_metric_snapshot_s3_bench.py | 74 ++++ .../tests/test_metric_snapshot_s3_ops.py | 203 +++++++++ .github/workflows/checks.yml | 2 + .../workflows/metric-snapshot-s3-bench.yml | 230 ++++++++++ docs/how-to/metric-snapshot-s3-bench.md | 79 ++++ 9 files changed, 1592 insertions(+) create mode 100644 .github/scripts/metric-snapshot-fixture.py create mode 100644 .github/scripts/metric-snapshot-s3-bench.py create mode 100644 .github/scripts/metric-snapshot-s3-ops.py create mode 100644 .github/scripts/tests/test_metric_snapshot_fixture.py create mode 100644 .github/scripts/tests/test_metric_snapshot_s3_bench.py create mode 100644 .github/scripts/tests/test_metric_snapshot_s3_ops.py create mode 100644 .github/workflows/metric-snapshot-s3-bench.yml create mode 100644 docs/how-to/metric-snapshot-s3-bench.md diff --git a/.github/scripts/metric-snapshot-fixture.py b/.github/scripts/metric-snapshot-fixture.py new file mode 100644 index 00000000000..4341bc6f579 --- /dev/null +++ b/.github/scripts/metric-snapshot-fixture.py @@ -0,0 +1,294 @@ +#!/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. + +"""Deterministic wide, heterogeneous Metric tables for snapshot benchmarks.""" + +import argparse +import base64 +import concurrent.futures +import hashlib +import http.client +import json +import os +from pathlib import Path +import threading +import time +import urllib.parse + +GROUPS = 10 +TAG_POOL = 298 +ROWS = 32 +DATABASE = "metrics" + + +class Client: + def __init__(self, addr): + self.addr = addr + self.connections = threading.local() + self.headers = {"Content-Type": "application/x-www-form-urlencoded"} + if auth := os.environ.get("GREPTIME_BENCH_AUTH"): + self.headers["Authorization"] = "Basic " + base64.b64encode( + auth.encode() + ).decode() + + def sql(self, query, database=DATABASE, batch=False): + connection = getattr(self.connections, "connection", None) + if connection is None: + connection = http.client.HTTPConnection(self.addr, timeout=600) + self.connections.connection = connection + body = urllib.parse.urlencode({"db": database, "sql": query}) + try: + connection.request( + "POST", "/v1/ddl/logical-tables" if batch else "/v1/sql", + body, self.headers, + ) + response = connection.getresponse() + raw = response.read() + except (OSError, http.client.HTTPException): + connection.close() + self.connections.connection = None + raise + data = json.loads(raw) + if response.status != 200 or "error" in data: + raise RuntimeError(f"SQL failed (HTTP {response.status}): {data}") + return data["output"] + + def values(self, query, database=DATABASE): + return self.sql(query, database)[0]["records"]["rows"] + + +def tags(index, per_group): + return sorted((index % per_group * 7 + offset) % TAG_POOL for offset in range(10)) + + +def column_names(tag_ids): + return ["ts", *[f"tag_{tag:03d}" for tag in tag_ids], "val"] + + +def columns(tag_ids): + names = [f"tag_{tag:03d}" for tag in tag_ids] + return ( + "ts TIMESTAMP TIME INDEX, " + + ", ".join(name + " STRING" for name in names) + + ", val DOUBLE, PRIMARY KEY(" + ",".join(names) + ")" + ) + + +def table_ddl(index, per_group): + return ( + f"CREATE TABLE m{index:06d} ({columns(tags(index, per_group))}) " + f"ENGINE=metric WITH(on_physical_table='phy_{index // per_group:02d}')" + ) + + +def row(index, row_index, per_group): + values = [] + for tag in tags(index, per_group): + if (row_index + tag) % 29 == 0: + value = None + elif (row_index + tag) % 17 == 0: + value = "" + elif tag % 10 == 0: + value = f"instance-{index:06d}-{row_index % 4}" + elif tag % 10 == 1: + value = "区域-" + str(row_index % 3) + elif tag % 10 == 2: + value = "label-" + "x" * 48 + "-" + str(row_index % 2) + else: + value = f"tag{tag:03d}-v{row_index % (2 + tag % 7)}" + values.append(value) + return [row_index, *values, None if row_index % 7 == 0 else index % 97 + row_index / 4] + + +def sql_value(value): + if value is None: + return "NULL" + if isinstance(value, str): + return "'" + value.replace("'", "''") + "'" + return str(value) + + +def canonical(value): + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def update_digest(digest, value): + digest.update((canonical(value) + "\n").encode()) + + +def progress(phase, completed, total): + if completed == total or completed % 4096 == 0: + print(f"{phase}: {completed}/{total}", flush=True) + + +def prepare(client, per_group): + stages = {} + started = time.monotonic() + client.sql(f"CREATE DATABASE {DATABASE}", "public") + for group in range(GROUPS): + client.sql( + f"CREATE TABLE phy_{group:02d} ({columns(range(TAG_POOL))}) " + "ENGINE=metric WITH(physical_metric_table='')" + ) + total = GROUPS * per_group + for group in range(GROUPS): + for base in range(group * per_group, (group + 1) * per_group, 128): + end = min((group + 1) * per_group, base + 128) + client.sql(";".join(table_ddl(i, per_group) for i in range(base, end)), batch=True) + progress("create", end, total) + stages["create_seconds"] = time.monotonic() - started + + def insert(index): + rows = ",".join( + "(" + ",".join(sql_value(v) for v in row(index, j, per_group)) + ")" + for j in range(ROWS) + ) + names = ",".join(column_names(tags(index, per_group))) + client.sql(f"INSERT INTO m{index:06d} ({names}) VALUES {rows}") + + started = time.monotonic() + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + for base in range(0, total, 128): + end = min(total, base + 128) + list(pool.map(insert, range(base, end))) + progress("insert", end, total) + stages["insert_seconds"] = time.monotonic() - started + started = time.monotonic() + for group in range(GROUPS): + client.sql(f"ADMIN FLUSH_TABLE('{DATABASE}.phy_{group:02d}')") + stages["flush_seconds"] = time.monotonic() - started + return stages + + +def check_records(index, per_group, records): + names = column_names(tags(index, per_group)) + types = ["TimestampMillisecond", *["String"] * 10, "Float64"] + expected_columns = [dict(name=name, data_type=kind) for name, kind in zip(names, types)] + actual_columns = records["schema"]["column_schemas"] + if sorted(actual_columns, key=lambda c: c["name"]) != sorted( + expected_columns, key=lambda c: c["name"] + ): + raise RuntimeError(f"logical schema mismatch: m{index:06d}") + # Metric tables canonicalize column order; SELECT * still checks every column. + positions = [names.index(column["name"]) for column in actual_columns] + expected_rows = [ + [values[position] for position in positions] + for values in (row(index, j, per_group) for j in range(ROWS)) + ] + # Canonical JSON also distinguishes integers, floats, booleans, and NULL. + if canonical(records["rows"]) != canonical(expected_rows): + raise RuntimeError(f"typed values/NULL mismatch: m{index:06d}") + + +def validate(client, per_group): + total = GROUPS * per_group + expected_names = {f"m{i:06d}" for i in range(total)} | { + f"phy_{group:02d}" for group in range(GROUPS) + } + table_rows = client.values( + "SELECT table_name, table_id FROM information_schema.tables " + f"WHERE table_schema='{DATABASE}' ORDER BY table_name" + ) + table_ids = dict(table_rows) + if len(table_rows) != len(expected_names) or set(table_ids) != expected_names: + raise RuntimeError("table membership mismatch") + if len(set(table_ids.values())) != len(table_ids): + raise RuntimeError("duplicate table IDs") + digests = {name: hashlib.sha256() for name in ("schema", "data", "ddl")} + + def check_ddl(names): + outputs = client.sql(";".join(f"SHOW CREATE TABLE {name}" for name in names)) + if len(outputs) != len(names): + raise RuntimeError("incomplete SHOW CREATE TABLE response") + for name, output in zip(names, outputs): + records = output["records"]["rows"] + if len(records) != 1 or records[0][0] != name or not records[0][1]: + raise RuntimeError(f"invalid SHOW CREATE TABLE response: {name}") + update_digest(digests["ddl"], records) + + for base in range(0, total, 64): + end = min(total, base + 64) + names = [f"m{i:06d}" for i in range(base, end)] + outputs = client.sql(";".join(f"SELECT * FROM {name} ORDER BY ts" for name in names)) + if len(outputs) != len(names): + raise RuntimeError("incomplete SELECT response") + for index, output in enumerate(outputs, base): + records = output["records"] + check_records(index, per_group, records) + update_digest(digests["schema"], [f"m{index:06d}", records["schema"]]) + update_digest(digests["data"], [f"m{index:06d}", records["rows"]]) + check_ddl(names) + progress("validate", end, total) + check_ddl([f"phy_{group:02d}" for group in range(GROUPS)]) + return {**{name + "_sha256": digest.hexdigest() for name, digest in digests.items()}, + "table_ids": table_ids} + + +def compare_source(result, source): + for field in ("fixture", "schema_sha256", "data_sha256", "ddl_sha256"): + if result[field] != source[field]: + raise RuntimeError(f"source/target mismatch: {field}") + source_ids, target_ids = source["table_ids"], result["table_ids"] + if source_ids.keys() != target_ids.keys(): + raise RuntimeError("source/target table membership mismatch") + changed = sum(source_ids[name] != target_ids[name] for name in source_ids) + if changed != len(source_ids): + raise RuntimeError(f"table IDs not all remapped: {changed}/{len(source_ids)}") + result["remapped_table_ids"] = changed + + +def placeholders(client): + database = "metric_bench_guard" + client.sql(f"CREATE DATABASE {database}", "public") + for index in range(17): + client.sql(f"CREATE TABLE placeholder_{index} (ts TIMESTAMP TIME INDEX)", database) + return {"placeholder_tables": 17} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("action", choices=("prepare", "validate", "placeholders")) + parser.add_argument("--addr", default="127.0.0.1:4000") + parser.add_argument("--tables-per-group", type=int, default=10000) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--source", type=Path, help="Source validation JSON, for target comparison") + args = parser.parse_args() + if not 50 <= args.tables_per_group <= 10000: + parser.error("--tables-per-group must be between 50 and 10000") + if args.source is not None and args.action != "validate": + parser.error("--source only applies to validate") + coverage = {tag for i in range(args.tables_per_group) for tag in tags(i, args.tables_per_group)} + if coverage != set(range(TAG_POOL)): + parser.error("fixture does not cover all 298 tags") + client = Client(args.addr) + started = time.monotonic() + result = {"fixture": {"groups": GROUPS, "tables_per_group": args.tables_per_group, + "rows_per_table": ROWS, "tag_pool": TAG_POOL}} + if args.action == "placeholders": + result.update(placeholders(client)) + elif args.action == "prepare": + result.update(prepare(client, args.tables_per_group)) + else: + result.update(validate(client, args.tables_per_group)) + result["wall_seconds"] = time.monotonic() - started + if args.source is not None: + compare_source(result, json.loads(args.source.read_text())) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n") + print(json.dumps({key: value for key, value in result.items() if key != "table_ids"})) + + +if __name__ == "__main__": + main() diff --git a/.github/scripts/metric-snapshot-s3-bench.py b/.github/scripts/metric-snapshot-s3-bench.py new file mode 100644 index 00000000000..7bb01a410b1 --- /dev/null +++ b/.github/scripts/metric-snapshot-s3-bench.py @@ -0,0 +1,392 @@ +#!/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. + +"""Run one packed Metric snapshot smoke, then one full S3 round trip.""" + +import argparse +from datetime import datetime +import json +import os +from pathlib import Path +import re +import signal +import subprocess +import sys +import threading +import time +import urllib.request + +HERE = Path(__file__).resolve().parent + + +def command(args, **kwargs): + result = subprocess.run(args, text=True, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, **kwargs) + if result.returncode: + # CLI argv and error output can contain storage credentials. + raise RuntimeError(f"{args[0]} exited with status {result.returncode}") + return result.stdout + + +def write_json(path, value): + path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n") + + +def metric_values(text): + values = {} + for line in text.splitlines(): + if line and not line.startswith("#"): + key, value = line.rsplit(" ", 1) + values[key] = float(value) + return values + + +def docker_timestamp(value): + # Docker emits nanoseconds; Python 3.9 accepts at most microseconds. + return datetime.fromisoformat(re.sub(r"(\.\d{6})\d+", r"\1", value).replace("Z", "+00:00")) + + +def host_memory(): + if not Path("/proc/vmstat").exists(): + return {"available": False} + vm = dict(line.split() for line in Path("/proc/vmstat").read_text().splitlines()) + mem = dict(line.split(":", 1) for line in Path("/proc/meminfo").read_text().splitlines()) + return {"available": True, "pswpin_pages": int(vm["pswpin"]), + "pswpout_pages": int(vm["pswpout"]), "oom_kill": int(vm.get("oom_kill", 0)), + **{key + "_kib": int(mem[key].split()[0]) + for key in ("MemTotal", "MemAvailable", "SwapTotal", "SwapFree")}} + + +def process_usage(pid): + try: + stat = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split() + return {"cpu_seconds": (int(stat[11]) + int(stat[12])) / os.sysconf("SC_CLK_TCK"), + "rss_bytes": int(stat[21]) * os.sysconf("SC_PAGE_SIZE")} + except (OSError, IndexError): + return None + + +class Benchmark: + def __init__(self, local_smoke): + self.local_smoke = local_smoke + self.image = os.environ["GREPTIMEDB_IMAGE"] + self.commit = os.environ["EXPECTED_COMMIT"] + if not re.fullmatch(r"greptime/greptimedb(?:-nightly)?@sha256:[0-9a-f]{64}", self.image): + raise ValueError("GREPTIMEDB_IMAGE must be an immutable official image digest") + if not re.fullmatch(r"[0-9a-f]{40}", self.commit): + raise ValueError("EXPECTED_COMMIT must be a full commit SHA") + self.per_group = int(os.environ.get("LOGICAL_TABLES_PER_GROUP", "10000")) + if not 50 <= self.per_group <= 10000: + raise ValueError("LOGICAL_TABLES_PER_GROUP must be in 50..10000") + self.prefix = os.environ["RUN_PREFIX"] + if not re.fullmatch(r"metric-snapshot-bench/[0-9]+-[0-9]+/", self.prefix): + raise ValueError("RUN_PREFIX must be metric-snapshot-bench/-/") + self.results = Path(os.environ["RESULTS"]).resolve() + self.data = Path(os.environ["DATA_ROOT"]).resolve() + self.results.mkdir(parents=True, exist_ok=False) + self.data.mkdir(parents=True, exist_ok=False) + self.network = "metric-snapshot-" + self.prefix.split("/")[1] + self.server = self.network + "-db" + self.cli_name = self.network + "-cli" + self.port = int(os.environ.get("BENCH_HTTP_PORT", "19410")) + self.base_url = f"http://127.0.0.1:{self.port}" + self.endpoint = os.environ.get("S3_ENDPOINT", "") + if self.endpoint and not local_smoke: + raise ValueError("Custom S3 endpoints are restricted to local smoke") + if not local_smoke and os.environ["S3_REGION"] != "ap-southeast-1": + raise ValueError("Cloud benchmark requires ap-southeast-1") + self.containers = set() + self.network_created = False + self.phases = [] + self.server_pid = 0 + self.round_dir = None + + def docker(self, *args): + return command(["docker", *args]) + + def metrics(self): + try: + with urllib.request.urlopen(self.base_url + "/metrics", timeout=5) as response: + return response.read().decode() + except OSError: + return "" + + def phase(self, name, action): + print(f"Starting {self.round_dir.name}/{name}", flush=True) + before = host_memory() + metrics_before = self.metrics() + samples = [] + done = threading.Event() + + def sample(): + while not done.is_set(): + samples.append({"time": time.time(), "server": process_usage(self.server_pid), + "cli": process_usage(getattr(self, "cli_pid", 0)), + "query_memory": {k: v for k, v in metric_values(self.metrics()).items() + if "query_memory" in k} if self.server_pid else {}}) + done.wait(0.5) + + monitor = threading.Thread(target=sample, daemon=True) + monitor.start() + start = time.monotonic() + outcome = "failed" + result = None + try: + result = action() + outcome = "passed" + return result + finally: + elapsed = time.monotonic() - start + done.set() + monitor.join() + after = host_memory() + interfered = before["available"] and after["available"] and ( + after["pswpout_pages"] > before["pswpout_pages"] or after["oom_kill"] > before["oom_kill"]) + if interfered: + outcome = "resource-invalid" + record = {"round": self.round_dir.name, "phase": name, "outcome": outcome, + "wall_seconds": elapsed, "host_before": before, "host_after": after, + "process_samples": samples} + if isinstance(result, dict): + record.update(result) + if before["available"] and after["available"]: + record["new_swapout_pages"] = after["pswpout_pages"] - before["pswpout_pages"] + record["new_oom_kills"] = after["oom_kill"] - before["oom_kill"] + metrics_after = self.metrics() + pre, post = metric_values(metrics_before), metric_values(metrics_after) + record["server_counter_deltas"] = { + key: value - pre.get(key, 0) for key, value in post.items() + if ("opendal" in key and ("_count{" in key or "_sum{" in key)) + or key.startswith("greptime_servers_http_requests_total{")} + write_json(self.round_dir / f"{name}-resources.json", record) + (self.round_dir / f"{name}-before.prom").write_text(metrics_before) + (self.round_dir / f"{name}-after.prom").write_text(metrics_after) + self.phases.append({k: v for k, v in record.items() if k != "process_samples"}) + self.summary() + print(f"Finished {name}: {outcome}, {elapsed:.3f}s", flush=True) + if interfered: + raise RuntimeError(f"Resource interference during {name}; evidence retained") + + def fixture(self, operation, per_group, output, source=None): + args = [sys.executable, str(HERE / "metric-snapshot-fixture.py"), operation, + "--addr", f"127.0.0.1:{self.port}", "--tables-per-group", str(per_group), + "--output", str(output)] + if source: + args += ["--source", str(source)] + with (self.round_dir / f"{operation}-{output.stem}.log").open("w") as log: + status = subprocess.run(args, stdout=log, stderr=subprocess.STDOUT).returncode + if status: + raise RuntimeError(f"Fixture {operation} failed; see its artifact log") + + def start_server(self, role): + root = self.data / self.round_dir.name / role + root.mkdir(parents=True) + evidence = self.round_dir / role + evidence.mkdir() + config = root / "config.toml" + config.write_text('''enable_telemetry = false +[http] +addr = "0.0.0.0:4000" +[grpc] +bind_addr = "0.0.0.0:4001" +[mysql] +enable = false +[postgres] +enable = false +[storage] +type = "File" +data_home = "/data/db" +[query] +parallelism = 4 +memory_pool_size = "2GiB" +[logging] +dir = "/evidence/logs" +level = "info" +''') + self.containers.add(self.server) + self.docker("run", "-d", "--name", self.server, "--network", self.network, + "--network-alias", "metric-db", "--user", f"{os.getuid()}:{os.getgid()}", + "--add-host", "host.docker.internal:host-gateway", + "-p", f"127.0.0.1:{self.port}:4000", + "-v", f"{root}:/data", "-v", f"{evidence}:/evidence", + self.image, "standalone", "start", "-c", "/data/config.toml") + self.server_pid = int(self.docker("inspect", "-f", "{{.State.Pid}}", self.server)) + deadline = time.monotonic() + 120 + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(self.base_url + "/v1/capabilities", timeout=2) as response: + caps = json.load(response) + if any(caps.get(key) != 1 for key in + ("metric_packed_export", "metric_packed_import", "metric_batch_ddl")): + raise RuntimeError("Required packed snapshot capabilities are absent") + write_json(evidence / "capabilities.json", caps) + return + except OSError: + if self.docker("inspect", "-f", "{{.State.Running}}", self.server).strip() != "true": + break + time.sleep(1) + raise RuntimeError("Server did not become ready") + + def stop_server(self, role): + self.docker("stop", "-t", "60", self.server) + state = json.loads(self.docker("inspect", "-f", "{{json .State}}", self.server)) + write_json(self.round_dir / role / "exit.json", state) + (self.round_dir / role / "stdout.log").write_text(self.docker("logs", self.server)) + self.docker("rm", self.server) + self.containers.remove(self.server) + self.server_pid = 0 + if state["OOMKilled"] or state["ExitCode"] != 0: + raise RuntimeError(f"{role} server did not shut down cleanly") + + def cli(self, args, name): + # Expand credentials only inside the container. Never persist argv or env. + script = 'exec greptime "$@" --s3 --s3-region "$S3_REGION" --s3-access-key-id "$AWS_ACCESS_KEY_ID" --s3-secret-access-key "$AWS_SECRET_ACCESS_KEY"' + if self.endpoint: + args += ["--s3-endpoint", self.endpoint] + self.containers.add(self.cli_name) + self.docker("create", "--name", self.cli_name, "--network", self.network, + "--user", f"{os.getuid()}:{os.getgid()}", + "--add-host", "host.docker.internal:host-gateway", + "-e", "AWS_ACCESS_KEY_ID", "-e", "AWS_SECRET_ACCESS_KEY", "-e", "S3_REGION", + "-v", f"{self.round_dir}:/evidence", "--entrypoint", "/bin/sh", self.image, + "-c", script, "sh", "--log-dir", f"/evidence/{name}-logs", "cli", "data", *args) + + def execute(): + self.docker("start", self.cli_name) + self.cli_pid = int(self.docker("inspect", "-f", "{{.State.Pid}}", self.cli_name)) + status = int(self.docker("wait", self.cli_name)) + state = json.loads(self.docker("inspect", "-f", "{{json .State}}", self.cli_name)) + (self.round_dir / f"{name}.log").write_text(self.docker("logs", self.cli_name)) + write_json(self.round_dir / f"{name}-exit.json", state) + if status or state["ExitCode"] or state["OOMKilled"]: + raise RuntimeError(f"{name} failed; see artifact log") + return {"cli_wall_seconds": (docker_timestamp(state["FinishedAt"]) - + docker_timestamp(state["StartedAt"])).total_seconds()} + + try: + self.phase(name, execute) + finally: + self.cli_pid = 0 + if not (self.round_dir / f"{name}-exit.json").exists(): + self.save_container(self.cli_name, name + "-failure") + self.docker("rm", "-f", self.cli_name) + self.containers.remove(self.cli_name) + + def round(self, label, per_group): + self.round_dir = self.results / label + self.round_dir.mkdir() + self.start_server("source") + source = self.round_dir / "source-validation.json" + self.phase("prepare", lambda: self.fixture("prepare", per_group, self.round_dir / "prepare.json")) + self.phase("source-validation", lambda: self.fixture("validate", per_group, source)) + location = f"s3://{os.environ['S3_BUCKET']}/{self.prefix}{label}" + self.cli(["export-v2", "create", "--addr", "metric-db:4000", "--to", location, + "--schemas", "metrics", "--pack-metric-data", "--parallelism", "4", + "--chunk-parallelism", "1", "--start-time", "1970-01-01T00:00:00Z", + "--end-time", "1970-01-01T00:00:01Z", "--timeout", "1h", + "--no-proxy", "--progress", "always"], "export") + if not self.local_smoke: + command(["uv", "run", str(HERE / "metric-snapshot-s3-ops.py"), "inventory", "--round", label, + "--output", str(self.round_dir / "s3-inventory.json")]) + self.stop_server("source") + self.cli(["export-v2", "verify", "--snapshot", location], "verify") + self.start_server("target") + self.fixture("placeholders", per_group, self.round_dir / "placeholders.json") + self.cli(["import-v2", "--addr", "metric-db:4000", "--from", location, + "--task-parallelism", "1", "--state-path", "/evidence/import-state.json", + "--timeout", "1h", "--no-proxy", "--progress", "always"], "import") + imported = json.loads((self.round_dir / "import-resources.json").read_text()) + batch_requests = sum(value for key, value in imported["server_counter_deltas"].items() + if '/v1/ddl/logical-tables' in key and 'code="200"' in key) + if batch_requests <= 0: + raise RuntimeError("Import did not complete any successful batch DDL HTTP requests") + write_json(self.round_dir / "import-completion.json", {"batch_ddl_requests": batch_requests, + "resume_state_removed": not (self.round_dir / "import-state.json").exists()}) + if (self.round_dir / "import-state.json").exists(): + raise RuntimeError("Successful import left its resume state file") + self.phase("target-validation", lambda: self.fixture( + "validate", per_group, self.round_dir / "target-validation.json", source)) + self.stop_server("target") + + def summary(self): + lines = ["# Metric packed snapshot benchmark", "", f"Image: `{self.image}`", + f"Source commit: `{self.commit}`", "", + "Local smoke only; timings are not AWS results." if self.local_smoke else + "AWS S3 in ap-southeast-1; source and target use local File storage.", "", + "| Round | Phase | Seconds | Result |", "| --- | --- | ---: | --- |"] + for phase in self.phases: + lines.append(f"| {phase['round']} | {phase['phase']} | {phase['wall_seconds']:.3f} | {phase['outcome']} |") + lines += ["", "Each logical table has 32 rows, 10 overlapping string tags, a timestamp and a DOUBLE.", + "Phase durations include Docker start, wait and log collection. cli_wall_seconds records Docker StartedAt to FinishedAt separately; setup, loading and full validation are separate.", + "Process CPU/RSS are sampled every 0.5 s on Linux, not exact process peaks. Linux host swap/OOM counters are checked per phase; configured swap is not RAM.", + "Raw server Prometheus snapshots and CLI progress/logs are retained. OpenDAL operation counts are not AWS HTTP request counts: multipart requests, retries and CLI operations are not fully covered. Exact AWS request totals are unavailable."] + (self.results / "summary.md").write_text("\n".join(lines) + "\n") + + def run(self): + self.docker("pull", self.image) + version = self.docker("run", "--rm", self.image, "--version") + (self.results / "version.txt").write_text(version) + if self.commit not in version or "clean: true" not in version: + raise RuntimeError("Image version does not match the expected clean source commit") + for args, required in ((["export-v2", "create"], "--pack-metric-data"), + (["import-v2"], "--task-parallelism")): + help_text = self.docker("run", "--rm", self.image, "cli", "data", *args, "--help") + (self.results / (args[0] + "-help.txt")).write_text(help_text) + if required not in help_text: + raise RuntimeError(f"Image CLI lacks {required}") + write_json(self.results / "host.json", {"memory": host_memory(), "local_smoke": self.local_smoke, + "platform": self.docker("image", "inspect", "-f", "{{.Os}}/{{.Architecture}}", self.image).strip()}) + self.docker("network", "create", self.network) + self.network_created = True + self.round("smoke", 50) + if not self.local_smoke and self.per_group != 50: + self.round("full", self.per_group) + + def save_container(self, name, label): + state = subprocess.run(["docker", "inspect", "-f", "{{json .State}}", name], + capture_output=True, text=True) + (self.round_dir / (label + "-state.json")).write_text(state.stdout) + result = subprocess.run(["docker", "logs", name], capture_output=True, text=True) + (self.round_dir / (label + ".log")).write_text(result.stdout + result.stderr) + + def cleanup(self): + for name in self.containers: + if self.round_dir: + self.save_container(name, name + "-failure") + subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, check=False) + if self.network_created: + subprocess.run(["docker", "network", "rm", self.network], stdout=subprocess.DEVNULL, check=False) + self.summary() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("command", choices=["run"]) + parser.add_argument("--local-smoke", action="store_true") + args = parser.parse_args() + bench = Benchmark(args.local_smoke) + + def cancelled(signum, frame): + raise RuntimeError(f"Received signal {signum}") + + signal.signal(signal.SIGTERM, cancelled) + try: + bench.run() + finally: + bench.cleanup() + + +if __name__ == "__main__": + main() diff --git a/.github/scripts/metric-snapshot-s3-ops.py b/.github/scripts/metric-snapshot-s3-ops.py new file mode 100644 index 00000000000..f1e34defb89 --- /dev/null +++ b/.github/scripts/metric-snapshot-s3-ops.py @@ -0,0 +1,217 @@ +#!/usr/bin/env python3 +# Copyright 2023 Greptime Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# /// script +# requires-python = ">=3.10" +# dependencies = ["boto3>=1.35,<2"] +# /// + +"""Budget preflight and strictly scoped S3 operations for the metric snapshot benchmark.""" + +import argparse +from decimal import Decimal +import json +import os +from pathlib import Path +import re +import time +import urllib.request + +REGION = "ap-southeast-1" +HOURLY_USD = Decimal("0.9408") +OTHER_RESERVE_USD = Decimal("1.50") + + +def owned_prefix(prefix): + if not re.fullmatch(r"metric-snapshot-bench/[0-9]+-[0-9]+/", prefix): + raise ValueError("prefix must be exactly metric-snapshot-bench/-/") + return prefix + + +def plan(env): + image = env["GREPTIMEDB_IMAGE"] + if not re.fullmatch(r"greptime/greptimedb(?:-nightly)?@sha256:[0-9a-f]{64}", image): + raise ValueError("image must be an immutable greptime/greptimedb sha256 reference") + if not re.fullmatch(r"[0-9a-f]{40}", env["EXPECTED_COMMIT"]): + raise ValueError("expected commit must be a full lowercase source SHA") + tables, minutes = int(env["LOGICAL_TABLES_PER_GROUP"]), int(env["BENCHMARK_TIMEOUT_MINUTES"]) + if not 50 <= tables <= 10_000 or not 30 <= minutes <= 240: + raise ValueError("logical tables per group must be 50..10000; timeout must be 30..240 minutes") + if env["S3_REGION"] != REGION or env["RUNNER_REGION"] != REGION: + raise ValueError("bucket and runner must both use ap-southeast-1") + if not env["S3_BUCKET"]: + raise ValueError("S3 bucket is required") + prior = Decimal(env["PRIOR_SPEND_USD"]) + # Reserve 45m provision, 10m queue allowance and 20m cleanup beyond the job timeout. + reservation = HOURLY_USD * Decimal(minutes + 75) / 60 + OTHER_RESERVE_USD + if not prior.is_finite() or prior < 0 or prior >= 15 or prior + reservation > 20: + raise ValueError("campaign budget exhausted: prior spend must be below $15 and leave the full attempt reserve within $20") + return {"image": image, "expected_commit": env["EXPECTED_COMMIT"], + "logical_tables_per_group": tables, "smoke_logical_tables": 500, + "timeout_minutes": minutes, "prefix": owned_prefix(env["RUN_PREFIX"]), + "instance_type": "c6id.4xlarge", "region": REGION, + "prior_estimated_usd": str(prior), "attempt_reserve_usd": str(reservation), + "ec2_hourly_usd": str(HOURLY_USD), "other_reserve_usd": str(OTHER_RESERVE_USD), + "cost_note": "Estimate, not an AWS billing total or a hard spending cap."} + + +def client(region): + import boto3 + from botocore.config import Config + + if region != REGION: + raise ValueError("S3 region must be ap-southeast-1") + return boto3.client("s3", region_name=region, config=Config( + connect_timeout=10, read_timeout=30, retries={"mode": "standard", "max_attempts": 3})) + + +def entries(s3, operation, bucket, prefix, fields, scope=""): + listed = owned_prefix(prefix) + scope + for page in s3.get_paginator(operation).paginate(Bucket=bucket, Prefix=listed): + for field in fields: + for item in page.get(field, []): + if not item["Key"].startswith(listed): + raise ValueError("S3 returned an object outside the owned prefix") + yield item + + +def inventory(s3, bucket, prefix, round=None): + # A round's snapshot lives under /; earlier rounds stay until cleanup. + if round not in (None, "smoke", "full"): + raise ValueError("round must be smoke or full") + scope = round + "/" if round else "" + objects = list(entries(s3, "list_objects_v2", bucket, prefix, ("Contents",), scope)) + return {"prefix": prefix + scope, "objects": len(objects), "bytes": sum(item["Size"] for item in objects), + "measurement": "S3 listing inventory; not request counts or transferred bytes."} + + +def probe(s3, bucket, prefix): + if inventory(s3, bucket, prefix)["objects"]: + raise ValueError("run prefix is already occupied; use a fresh workflow attempt") + key = prefix + "_probe" + s3.put_object(Bucket=bucket, Key=key, Body=b"metric snapshot benchmark") + try: + body = s3.get_object(Bucket=bucket, Key=key)["Body"] + try: + if body.read() != b"metric snapshot benchmark": + raise ValueError("S3 probe returned unexpected content") + finally: + body.close() + finally: + s3.delete_object(Bucket=bucket, Key=key) + return inventory(s3, bucket, prefix) + + +def cleanup(s3, bucket, prefix): + owned_prefix(prefix) + for item in entries(s3, "list_objects_v2", bucket, prefix, ("Contents",)): + s3.delete_object(Bucket=bucket, Key=item["Key"]) + after = inventory(s3, bucket, prefix) + if after["objects"]: + raise RuntimeError("owned S3 prefix is not empty after cleanup") + return after + + +def scrub(root, env): + secrets = [env[name].encode() for name in ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN") + if env.get(name)] + if not secrets: + raise ValueError("credential scan requires the CI credential environment") + overlap = max(map(len, secrets)) - 1 + for result in Path(root).rglob("*"): + if result.is_file(): + with result.open("rb") as stream: + previous = b"" + while chunk := stream.read(1024 * 1024): + data = previous + chunk + if any(secret in data for secret in secrets): + raise ValueError("credential detected in results; artifact upload is blocked") + previous = data[-overlap:] if overlap else b"" + + +def wait_for_benchmark(env, started_at): + """Bound runner allocation even when the benchmark job never leaves its queue.""" + run_id, attempt = owned_prefix(env["RUN_PREFIX"]).split("/")[1].split("-") + repo = env["GITHUB_REPOSITORY"] + if run_id != env["GITHUB_RUN_ID"] or not re.fullmatch(r"[\w.-]+/[\w.-]+", repo): + raise ValueError("workflow identity does not match its owned resource prefix") + minutes = int(env["BENCHMARK_TIMEOUT_MINUTES"]) + if not 30 <= minutes <= 240 or started_at <= 0: + raise ValueError("invalid benchmark allocation deadline") + # Use the original allocation timestamp; a delayed or retried watcher cannot extend it. + deadline = started_at + (45 + minutes + 10) * 60 + request = urllib.request.Request( + f"https://api.github.com/repos/{repo}/actions/runs/{run_id}/attempts/{attempt}/jobs?per_page=100", + headers={"Authorization": f"Bearer {env['GH_TOKEN']}", "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28"}) + while True: + remaining = deadline - time.time() + if remaining <= 0: + raise TimeoutError("benchmark allocation deadline reached; proceeding to resource cleanup") + # API errors deliberately fail this step: subsequent always() steps clean up immediately. + with urllib.request.urlopen(request, timeout=min(15, remaining)) as response: + jobs = json.load(response)["jobs"] + benchmark = next((job for job in jobs if job["name"] == "benchmark"), None) + if benchmark and benchmark["status"] == "completed": + return {"benchmark_conclusion": benchmark["conclusion"], "owned_attempt": attempt} + remaining = deadline - time.time() + if remaining > 0: + time.sleep(min(30, remaining)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("command", choices=("plan", "probe", "inventory", "cleanup", "scrub", "cost", "wait-for-benchmark")) + parser.add_argument("--output") + parser.add_argument("--started-at", type=int) + parser.add_argument("--round") + args = parser.parse_args() + if args.round is not None and args.command != "inventory": + parser.error("--round applies only to inventory") + if args.command == "plan": + result = plan(os.environ) + elif args.command == "scrub": + scrub(os.environ["RESULTS"], os.environ) + return + elif args.command == "wait-for-benchmark": + if args.started_at is None: + parser.error("wait-for-benchmark requires --started-at") + result = wait_for_benchmark(os.environ, args.started_at) + elif args.command == "cost": + if args.started_at is None: + parser.error("cost requires --started-at") + seconds = max(0, int(time.time()) - args.started_at) + estimate = HOURLY_USD * Decimal(seconds) / 3600 + result = {"allocation_window_seconds": seconds, "ec2_estimate_usd": str(round(estimate, 4)), + "campaign_estimate_with_other_reserve_usd": str(round( + Decimal(os.environ["PRIOR_SPEND_USD"]) + estimate + OTHER_RESERVE_USD, 4)), + "note": "Allocation wall-time estimate plus $1.50 reserve for S3, disk and IPv4; not an AWS bill."} + else: + prefix = owned_prefix(os.environ["RUN_PREFIX"]) + s3, bucket = client(os.environ["S3_REGION"]), os.environ["S3_BUCKET"] + if args.command == "inventory": + result = inventory(s3, bucket, prefix, args.round) + else: + result = globals()[args.command](s3, bucket, prefix) + rendered = json.dumps(result, indent=2) + "\n" + if args.output: + destination = Path(args.output) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(rendered) + print(rendered, end="") + + +if __name__ == "__main__": + main() diff --git a/.github/scripts/tests/test_metric_snapshot_fixture.py b/.github/scripts/tests/test_metric_snapshot_fixture.py new file mode 100644 index 00000000000..f4c16a11af3 --- /dev/null +++ b/.github/scripts/tests/test_metric_snapshot_fixture.py @@ -0,0 +1,101 @@ +# 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. + +import copy +import importlib.util +from pathlib import Path +import unittest + +spec = importlib.util.spec_from_file_location( + "fixture", Path(__file__).resolve().parents[1] / "metric-snapshot-fixture.py" +) +fixture = importlib.util.module_from_spec(spec) +spec.loader.exec_module(fixture) + + +class ValidatorTests(unittest.TestCase): + def records(self): + return { + "schema": {"column_schemas": [ + {"name": "ts", "data_type": "TimestampMillisecond"}, + *[{"name": f"tag_{i:03d}", "data_type": "String"} for i in range(10)], + {"name": "val", "data_type": "Float64"}, + ]}, + "rows": [fixture.row(0, j, 50) for j in range(32)], + } + + def test_full_tag_coverage_and_heterogeneous_schemas(self): + for per_group in (50, 10000): + variants = {tuple(fixture.tags(i, per_group)) for i in range(per_group)} + self.assertEqual(set().union(*map(set, variants)), set(range(298))) + self.assertGreater(len(variants), 1) + self.assertTrue(all(len(tags) == 10 for tags in variants)) + + def test_accepts_engine_column_order_with_matching_values(self): + records = self.records() + order = [*range(1, 11), 0, 11] # The engine returns tags, timestamp, value. + records["schema"]["column_schemas"] = [ + records["schema"]["column_schemas"][index] for index in order + ] + records["rows"] = [[values[index] for index in order] for values in records["rows"]] + fixture.check_records(0, 50, records) + records["rows"][1][10] = 2 + with self.assertRaisesRegex(RuntimeError, "typed values"): + fixture.check_records(0, 50, records) + + def test_rejects_column_type_and_membership_changes(self): + fixture.check_records(0, 50, self.records()) + for mutation in ("type", "extra", "missing"): + records = self.records() + columns = records["schema"]["column_schemas"] + if mutation == "type": + columns[0]["data_type"] = "TimestampMicrosecond" + elif mutation == "extra": + columns.append({"name": "unexpected", "data_type": "String"}) + else: + del columns[1] + with self.subTest(mutation=mutation), self.assertRaisesRegex(RuntimeError, "schema"): + fixture.check_records(0, 50, records) + + def test_rejects_null_and_numeric_type_changes(self): + for value in ("", 0, False): + records = self.records() + records["rows"][0][-1] = value # Expected SQL NULL. + with self.subTest(value=value), self.assertRaisesRegex(RuntimeError, "typed values"): + fixture.check_records(0, 50, records) + records = self.records() + records["rows"][4][-1] = 1 # Numeric equality to 1.0 must not hide a type change. + with self.assertRaisesRegex(RuntimeError, "typed values"): + fixture.check_records(0, 50, records) + + def test_source_comparison_requires_same_ddl_and_remapped_ids(self): + source = {"fixture": {"tables_per_group": 50}, "schema_sha256": "schema", + "data_sha256": "data", "ddl_sha256": "ddl", + "table_ids": {"m000000": 1024, "phy_00": 1025}} + target = copy.deepcopy(source) + target["table_ids"] = {"m000000": 1041, "phy_00": 1042} + fixture.compare_source(target, source) + self.assertEqual(target["remapped_table_ids"], 2) + for field in ("schema_sha256", "data_sha256", "ddl_sha256"): + changed = copy.deepcopy(target) + changed[field] = "changed" + with self.subTest(field=field), self.assertRaisesRegex(RuntimeError, field): + fixture.compare_source(changed, source) + target["table_ids"]["phy_00"] = 1025 + with self.assertRaisesRegex(RuntimeError, "not all remapped"): + fixture.compare_source(target, source) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/tests/test_metric_snapshot_s3_bench.py b/.github/scripts/tests/test_metric_snapshot_s3_bench.py new file mode 100644 index 00000000000..c662ecbff76 --- /dev/null +++ b/.github/scripts/tests/test_metric_snapshot_s3_bench.py @@ -0,0 +1,74 @@ +#!/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. + +"""Failure accounting for the metric snapshot runner.""" +import importlib.util +import json +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +SPEC = importlib.util.spec_from_file_location( + "metric_snapshot_bench", Path(__file__).resolve().parents[1] / "metric-snapshot-s3-bench.py") +bench = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(bench) + + +class PhaseTests(unittest.TestCase): + def test_docker_nanosecond_timestamps(self): + start = bench.docker_timestamp("2026-10-14T06:06:44.72000818Z") + end = bench.docker_timestamp("2026-10-14T06:06:46.123456789Z") + self.assertAlmostEqual((end - start).total_seconds(), 1.403448) + + def test_resource_interference_invalidates_success_and_preserves_fixture(self): + with tempfile.TemporaryDirectory() as directory: + runner = bench.Benchmark.__new__(bench.Benchmark) + runner.round_dir = Path(directory) + runner.server_pid = 0 + runner.phases = [] + runner.metrics = lambda: "" + runner.summary = lambda: None + fixture = runner.round_dir / "source-validation.json" + fixture.write_text('{"data_sha256":"fixture"}') + before = {"available": True, "pswpout_pages": 0, "oom_kill": 0} + after = {"available": True, "pswpout_pages": 1, "oom_kill": 0} + with patch.object(bench, "host_memory", side_effect=[before, after]): + with self.assertRaisesRegex(RuntimeError, "Resource interference"): + runner.phase("source-validation", lambda: None) + self.assertEqual(json.loads(fixture.read_text()), {"data_sha256": "fixture"}) + evidence = json.loads((runner.round_dir / "source-validation-resources.json").read_text()) + self.assertEqual(evidence["outcome"], "resource-invalid") + self.assertEqual(evidence["new_swapout_pages"], 1) + + def test_action_failure_keeps_failed_resource_evidence(self): + with tempfile.TemporaryDirectory() as directory: + runner = bench.Benchmark.__new__(bench.Benchmark) + runner.round_dir = Path(directory) + runner.server_pid = 0 + runner.phases = [] + runner.metrics = lambda: "" + runner.summary = lambda: None + def fail(): + raise RuntimeError("export failed") + with patch.object(bench, "host_memory", return_value={"available": False}): + with self.assertRaisesRegex(RuntimeError, "export failed"): + runner.phase("export", fail) + evidence = json.loads((runner.round_dir / "export-resources.json").read_text()) + self.assertEqual(evidence["outcome"], "failed") + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/tests/test_metric_snapshot_s3_ops.py b/.github/scripts/tests/test_metric_snapshot_s3_ops.py new file mode 100644 index 00000000000..d22c09044b2 --- /dev/null +++ b/.github/scripts/tests/test_metric_snapshot_s3_ops.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +# Copyright 2023 Greptime Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Offline safety checks for the metric snapshot workflow's budget and cleanup.""" + +import importlib.util +import io +import json +from pathlib import Path +import tempfile +import unittest +from unittest import mock + +SPEC = importlib.util.spec_from_file_location("snapshot_ops", Path(__file__).parents[1] / "metric-snapshot-s3-ops.py") +ops = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(ops) +PREFIX = "metric-snapshot-bench/123-1/" + + +class FakeS3: + def __init__(self): + self.objects = [{"Key": PREFIX + "current", "Size": 11}, {"Key": PREFIX + "data", "Size": 7}] + self.deleted = [] + + def get_paginator(self, operation): + assert operation == "list_objects_v2" + parent = self + + class Paginator: + def paginate(self, **kwargs): + assert kwargs["Bucket"] == "bucket" and kwargs["Prefix"].startswith(PREFIX) + # Keys outside the run prefix model a misbehaving listing and are always returned. + return iter([{"Contents": [item for item in parent.objects if not item["Key"].startswith(PREFIX) + or item["Key"].startswith(kwargs["Prefix"])]}]) + + return Paginator() + + def put_object(self, **kwargs): + self.objects.append({"Key": kwargs["Key"], "Size": len(kwargs["Body"])}) + return {} + + def get_object(self, **kwargs): + return {"Body": io.BytesIO(b"metric snapshot benchmark")} + + def delete_object(self, **kwargs): + self.deleted.append(kwargs) + self.objects = [item for item in self.objects if item["Key"] != kwargs["Key"]] + + +class OperationsTest(unittest.TestCase): + def env(self, **changes): + return dict(GREPTIMEDB_IMAGE="greptime/greptimedb-nightly@sha256:" + "a" * 64, + EXPECTED_COMMIT="b" * 40, LOGICAL_TABLES_PER_GROUP="10000", + BENCHMARK_TIMEOUT_MINUTES="240", PRIOR_SPEND_USD="0", + S3_REGION="ap-southeast-1", RUNNER_REGION="ap-southeast-1", S3_BUCKET="bucket", + RUN_PREFIX=PREFIX) | changes + + def test_budget_includes_provision_cleanup_and_other_cost_reserve(self): + plan = ops.plan(self.env()) + self.assertEqual(plan["attempt_reserve_usd"], "6.4392") + ops.plan(self.env(PRIOR_SPEND_USD="13.5608")) + for changes in ({"PRIOR_SPEND_USD": "13.5609"}, {"PRIOR_SPEND_USD": "15"}, + {"PRIOR_SPEND_USD": "NaN"}, {"PRIOR_SPEND_USD": "-1"}, + {"BENCHMARK_TIMEOUT_MINUTES": "241"}, {"LOGICAL_TABLES_PER_GROUP": "49"}, + {"LOGICAL_TABLES_PER_GROUP": "10001"}, {"S3_REGION": "us-east-1"}, + {"GREPTIMEDB_IMAGE": "greptime/greptimedb:latest"}, + {"GREPTIMEDB_IMAGE": "docker.io/greptime/greptimedb@sha256:" + "a" * 64}, + {"GREPTIMEDB_IMAGE": "greptime/greptimedb:nightly@sha256:" + "a" * 64}, + {"EXPECTED_COMMIT": "abcdef"}): + with self.subTest(changes=changes), self.assertRaises(ValueError): + ops.plan(self.env(**changes)) + + def watcher_env(self): + return self.env(GITHUB_REPOSITORY="GreptimeTeam/greptimedb", GITHUB_RUN_ID="123", + GITHUB_RUN_ATTEMPT="2", GH_TOKEN="test-token", BENCHMARK_TIMEOUT_MINUTES="30") + + def test_watcher_completed_job_uses_original_attempt_and_does_not_wait(self): + for conclusion in ("success", "failure", "cancelled"): + response = io.BytesIO(json.dumps({"jobs": [{"name": "benchmark", "status": "completed", + "conclusion": conclusion}]}).encode()) + with self.subTest(conclusion=conclusion), \ + mock.patch.object(ops.urllib.request, "urlopen", return_value=response) as fetch, \ + mock.patch.object(ops.time, "time", return_value=1005), \ + mock.patch.object(ops.time, "sleep") as sleep: + self.assertEqual(ops.wait_for_benchmark(self.watcher_env(), 1000), + {"benchmark_conclusion": conclusion, "owned_attempt": "1"}) + self.assertIn("/runs/123/attempts/1/jobs?", fetch.call_args.args[0].full_url) + sleep.assert_not_called() + + def test_watcher_queue_reaches_durable_deadline(self): + deadline = 1000 + (45 + 30 + 10) * 60 + clock = [deadline - 31] + + def advance(seconds): + self.assertLessEqual(seconds, 30) + clock[0] += seconds + + def queued(*args, **kwargs): + return io.BytesIO(b'{"jobs": [{"name": "benchmark", "status": "queued"}]}') + + with mock.patch.object(ops.urllib.request, "urlopen", side_effect=queued) as fetch, \ + mock.patch.object(ops.time, "time", side_effect=lambda: clock[0]), \ + mock.patch.object(ops.time, "sleep", side_effect=advance), \ + self.assertRaisesRegex(TimeoutError, "proceeding to resource cleanup"): + ops.wait_for_benchmark(self.watcher_env(), 1000) + self.assertEqual(clock[0], deadline) + self.assertEqual(fetch.call_count, 2) + + def test_watcher_api_error_fails_immediately_for_cleanup(self): + with mock.patch.object(ops.urllib.request, "urlopen", side_effect=OSError("API unavailable")), \ + mock.patch.object(ops.time, "time", return_value=1005), \ + mock.patch.object(ops.time, "sleep") as sleep, \ + self.assertRaises(OSError): + ops.wait_for_benchmark(self.watcher_env(), 1000) + sleep.assert_not_called() + + def test_cleanup_deletes_every_listed_object(self): + s3 = FakeS3() + result = ops.cleanup(s3, "bucket", PREFIX) + self.assertEqual(result["objects"], 0) + self.assertEqual(s3.deleted, [{"Bucket": "bucket", "Key": PREFIX + "current"}, + {"Bucket": "bucket", "Key": PREFIX + "data"}]) + + def test_probe_refuses_occupied_prefix_and_removes_its_object(self): + with self.assertRaisesRegex(ValueError, "already occupied"): + ops.probe(FakeS3(), "bucket", PREFIX) + s3 = FakeS3() + s3.objects = [] + self.assertEqual(ops.probe(s3, "bucket", PREFIX)["objects"], 0) + self.assertEqual(s3.deleted, [{"Bucket": "bucket", "Key": PREFIX + "_probe"}]) + + def test_failed_probe_delete_still_has_hosted_cleanup_path(self): + s3 = FakeS3() + s3.objects = [] + with mock.patch.object(s3, "delete_object", side_effect=OSError("temporary delete failure")), \ + self.assertRaises(OSError): + ops.probe(s3, "bucket", PREFIX) + self.assertEqual([item["Key"] for item in s3.objects], [PREFIX + "_probe"]) + workflow = (Path(__file__).parents[2] / "workflows/metric-snapshot-s3-bench.yml").read_text() + teardown = workflow.split(" teardown:\n", 1)[1] + job_if = next(line.strip() for line in teardown.splitlines() if line.startswith(" if:")) + self.assertEqual(job_if, "if: ${{ always() && needs.validate.outputs.run_id != '' }}") + ec2 = teardown.split(" - name: Delete EC2, its temporary disks and runner registration\n", 1)[1] + self.assertEqual(ec2.splitlines()[0].strip(), "if: ${{ always() && needs.provision.result != 'skipped' }}") + self.assertIn(" RUN_PREFIX: ${{ needs.validate.outputs.run_prefix }}\n", teardown.split(" steps:", 1)[0]) + s3_cleanup = teardown.split(" - name: Remove and verify only this run's S3 prefix\n", 1)[1].split(" - ", 1)[0] + self.assertEqual(s3_cleanup.splitlines()[0].strip(), "if: always()") + self.assertIn("run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py cleanup\n", s3_cleanup) + self.assertEqual(ops.cleanup(s3, "bucket", PREFIX)["objects"], 0) + + def test_round_inventory_excludes_other_rounds_while_cleanup_covers_the_run(self): + s3 = FakeS3() + s3.objects = [{"Key": PREFIX + "smoke/pack-000000.bin", "Size": 3}, + {"Key": PREFIX + "full/pack-000000.bin", "Size": 5}, + {"Key": PREFIX + "full/index.json", "Size": 2}] + full = ops.inventory(s3, "bucket", PREFIX, "full") + self.assertEqual((full["prefix"], full["objects"], full["bytes"]), (PREFIX + "full/", 2, 7)) + self.assertEqual(ops.inventory(s3, "bucket", PREFIX, "smoke")["bytes"], 3) + self.assertEqual(ops.inventory(s3, "bucket", PREFIX)["objects"], 3) + for round in ("", "../full", "full/"): + with self.subTest(round=round), self.assertRaises(ValueError): + ops.inventory(s3, "bucket", PREFIX, round) + self.assertEqual(ops.cleanup(s3, "bucket", PREFIX)["objects"], 0) + self.assertEqual(len(s3.deleted), 3) + + def test_bad_prefix_and_out_of_scope_response_cannot_delete(self): + for prefix in ("", "metric-snapshot-bench/", "metric-snapshot-bench/123-1", PREFIX + "../", "oswal-bench/123-1/"): + s3 = FakeS3() + with self.subTest(prefix=prefix), self.assertRaises(ValueError): + ops.cleanup(s3, "bucket", prefix) + self.assertEqual(s3.deleted, []) + s3 = FakeS3() + s3.objects[0]["Key"] = "other-run/current" + with self.assertRaises(ValueError): + ops.inventory(s3, "bucket", PREFIX, "full") + with self.assertRaises(ValueError): + ops.cleanup(s3, "bucket", PREFIX) + self.assertEqual(s3.deleted, []) + + def test_credential_crossing_read_boundary_blocks_artifact_upload(self): + with tempfile.TemporaryDirectory() as directory: + logfile = Path(directory) / "server.log" + logfile.write_bytes(b"a" * (1024 * 1024 - 4) + b"private-key" + b"tail") + with self.assertRaisesRegex(ValueError, "upload is blocked"): + ops.scrub(directory, {"AWS_SECRET_ACCESS_KEY": "private-key"}) + logfile.write_text("safe log\n") + ops.scrub(directory, {"AWS_SECRET_ACCESS_KEY": "private-key"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index deba5010c80..2aa460ec0ca 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -88,6 +88,8 @@ jobs: run: .github/scripts/run-fuzz-targets-test.sh - name: Run CI comment-command tests run: python3 .github/scripts/tests/test_ci_slash.py + - name: Run metric snapshot benchmark tests + run: python3 -m unittest discover -s .github/scripts/tests -p 'test_metric_snapshot*.py' - name: Run query-regression Python tooling tests run: | python3 tests/perf/test_query_regression_runner_compaction_toctou.py diff --git a/.github/workflows/metric-snapshot-s3-bench.yml b/.github/workflows/metric-snapshot-s3-bench.yml new file mode 100644 index 00000000000..e4a12dde69f --- /dev/null +++ b/.github/workflows/metric-snapshot-s3-bench.yml @@ -0,0 +1,230 @@ +name: Metric Snapshot S3 Benchmark +on: + workflow_dispatch: + inputs: + greptimedb_image: + description: 'Immutable greptime/greptimedb or greptimedb-nightly image (repository@sha256:digest)' + required: true + type: string + expected_commit: + description: 'Full source commit SHA reported by the image' + required: true + type: string + logical_tables_per_group: + description: 'Logical tables per physical table, 50..10000; 10 physical tables, 32 rows per logical table' + type: number + default: 10000 + benchmark_timeout_minutes: + description: 'Setup, 500-table smoke and full run timeout, 30..240 minutes' + type: number + default: 240 + prior_spend_usd: + description: 'Cumulative estimated cloud spend for all earlier attempts in this $20 campaign' + required: true + type: string + default: '0' +permissions: + contents: read +concurrency: + group: metric-snapshot-s3-bench + cancel-in-progress: false +env: + GREPTIMEDB_IMAGE: ${{ inputs.greptimedb_image }} + EXPECTED_COMMIT: ${{ inputs.expected_commit }} + LOGICAL_TABLES_PER_GROUP: ${{ inputs.logical_tables_per_group }} + BENCHMARK_TIMEOUT_MINUTES: ${{ inputs.benchmark_timeout_minutes }} + PRIOR_SPEND_USD: ${{ inputs.prior_spend_usd }} + RUN_PREFIX: metric-snapshot-bench/${{ github.run_id }}-${{ github.run_attempt }}/ + S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }} + S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + RUNNER_REGION: ${{ vars.EC2_RUNNER_REGION }} +jobs: + validate: + runs-on: ubuntu-latest + timeout-minutes: 15 + outputs: + instance_type: ${{ steps.runner.outputs.instance_type }} + run_id: ${{ steps.identity.outputs.run_id }} + run_prefix: ${{ steps.identity.outputs.run_prefix }} + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - name: Test benchmark boundaries + run: | + python3 greptimedb/.github/scripts/tests/test_metric_snapshot_s3_ops.py + python3 greptimedb/.github/scripts/tests/test_metric_snapshot_fixture.py + python3 greptimedb/.github/scripts/tests/test_metric_snapshot_s3_bench.py + - name: Validate pinned version, scale, region and cumulative budget + run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py plan + - name: Bind resource ownership to this validated attempt + id: identity + run: | + echo "run_id=$GITHUB_RUN_ID-$GITHUB_RUN_ATTEMPT" >> "$GITHUB_OUTPUT" + echo "run_prefix=$RUN_PREFIX" >> "$GITHUB_OUTPUT" + - uses: astral-sh/setup-uv@v6 + - name: Check S3 access before allocating EC2 + env: + AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} + run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py probe + - name: Resolve the fixed benchmark runner + id: runner + run: python3 greptimedb/.github/scripts/ci-runner-config.py --provider AWS --instance-type c6id.4xlarge + provision: + needs: validate + runs-on: ubuntu-latest + timeout-minutes: 45 + outputs: + label: ${{ steps.aws.outputs.label }} + instance_id: ${{ steps.aws.outputs.instance_id }} + runner_name: ${{ steps.aws.outputs.runner_name }} + started_at: ${{ steps.start.outputs.started_at }} + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - uses: aws-actions/configure-aws-credentials@v4 + with: + aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + aws-region: ${{ env.RUNNER_REGION }} + - uses: astral-sh/setup-uv@v6 + - name: Record allocation start + id: start + run: echo "started_at=$(date +%s)" >> "$GITHUB_OUTPUT" + - name: Create ephemeral EC2 runner + id: aws + env: + GH_PERSONAL_ACCESS_TOKEN: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }} + AWS_REGION: ${{ env.RUNNER_REGION }} + AWS_EC2_IMAGE_ID: ${{ vars.BENCHMARK_EC2_IMAGE_ID || 'auto' }} + AWS_EC2_INSTANCE_TYPE: ${{ needs.validate.outputs.instance_type }} + AWS_EC2_SUBNET_ID: ${{ vars.EC2_RUNNER_SUBNET_ID }} + AWS_EC2_SECURITY_GROUP_ID: ${{ vars.EC2_RUNNER_SECURITY_GROUP_ID }} + AWS_EC2_SYSTEM_DISK_GIB: '60' + AWS_EC2_TTL_HOURS: '6' + AWS_EC2_RUN_ID: ${{ needs.validate.outputs.run_id }} + run: uv run greptimedb/.github/scripts/aws-ec2-runner-provision.py + benchmark: + needs: + - validate + - provision + runs-on: ${{ needs.provision.outputs.label }} + timeout-minutes: ${{ fromJSON(format('{0}', inputs.benchmark_timeout_minutes)) }} + env: + RUN_PREFIX: ${{ needs.validate.outputs.run_prefix }} + RESULTS: ${{ github.workspace }}/metric-snapshot-results + DISK_ROOT: /mnt/metric-snapshot-disk + DATA_ROOT: /mnt/metric-snapshot-disk/data + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - name: Set up benchmark dependencies + uses: ./greptimedb/.github/actions/setup-benchmark + - uses: astral-sh/setup-uv@v6 + - name: Mount the temporary local File data disk + run: | + device=$(lsblk -dnpo NAME,MODEL | awk '/Instance Storage/ {print $1; exit}') + test -n "$device" + if [[ "$(lsblk -nro TYPE "$device" | wc -l)" -ne 1 ]] || findmnt --source "$device" >/dev/null; then + echo 'Refusing partitioned or mounted instance storage' >&2 + exit 1 + fi + probe_status=0 + sudo -n blkid -p "$device" || probe_status=$? + if ((probe_status != 2)); then + echo 'Refusing a disk with an existing signature or an unsuccessful signature probe' >&2 + exit 1 + fi + sudo -n mkfs.ext4 -q "$device" + sudo -n mkdir -p "$DISK_ROOT" + sudo -n mount -o noatime "$device" "$DISK_ROOT" + sudo -n chown "$(id -u):$(id -g)" "$DISK_ROOT" + findmnt -no SOURCE,FSTYPE -T "$DISK_ROOT" + - name: Probe the real same-region S3 bucket + env: + AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} + run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py probe + - name: Run the 500-table smoke, then one full packed export and restore + env: + AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} + run: python3 greptimedb/.github/scripts/metric-snapshot-s3-bench.py run + - name: Check results for credentials + id: scrub + if: always() + env: + AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} + run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py scrub + - name: Publish the readable result + if: ${{ always() && steps.scrub.outcome == 'success' }} + run: | + if test -f "$RESULTS/summary.md"; then + cat "$RESULTS/summary.md" >> "$GITHUB_STEP_SUMMARY" + fi + - name: Upload bounded-retention evidence + if: ${{ always() && steps.scrub.outcome == 'success' }} + uses: actions/upload-artifact@v4 + with: + name: metric-snapshot-s3-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ env.RESULTS }} + if-no-files-found: warn + retention-days: 14 + teardown: + needs: + - validate + - provision + # Start alongside the benchmark so an offline runner cannot hold cleanup in its queue. + if: ${{ always() && needs.validate.outputs.run_id != '' }} + runs-on: ubuntu-latest + timeout-minutes: 330 + permissions: + contents: read + actions: read + env: + RUN_PREFIX: ${{ needs.validate.outputs.run_prefix }} + steps: + - uses: actions/checkout@v4 + with: + path: greptimedb + persist-credentials: false + - uses: astral-sh/setup-uv@v6 + - name: Wait for completion within the allocation deadline + if: ${{ !cancelled() && needs.provision.result == 'success' && needs.provision.outputs.started_at != '' }} + env: + GH_TOKEN: ${{ github.token }} + STARTED_AT: ${{ needs.provision.outputs.started_at }} + run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py wait-for-benchmark --started-at "$STARTED_AT" + - uses: aws-actions/configure-aws-credentials@v4 + if: ${{ always() && needs.provision.result != 'skipped' }} + with: + aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + aws-region: ${{ env.RUNNER_REGION }} + - name: Delete EC2, its temporary disks and runner registration + if: ${{ always() && needs.provision.result != 'skipped' }} + env: + GH_PERSONAL_ACCESS_TOKEN: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }} + AWS_REGION: ${{ env.RUNNER_REGION }} + AWS_EC2_INSTANCE_ID: ${{ needs.provision.outputs.instance_id }} + AWS_EC2_RUNNER_NAME: ${{ needs.provision.outputs.runner_name }} + AWS_EC2_RUN_ID: ${{ needs.validate.outputs.run_id }} + run: uv run greptimedb/.github/scripts/aws-ec2-runner-teardown.py + - name: Remove and verify only this run's S3 prefix + if: always() + env: + AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} + run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py cleanup + - name: Estimate the cumulative cloud cost + if: ${{ always() && needs.provision.outputs.started_at != '' }} + env: + STARTED_AT: ${{ needs.provision.outputs.started_at }} + run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py cost --started-at "$STARTED_AT" >> "$GITHUB_STEP_SUMMARY" diff --git a/docs/how-to/metric-snapshot-s3-bench.md b/docs/how-to/metric-snapshot-s3-bench.md new file mode 100644 index 00000000000..86468bf3ffd --- /dev/null +++ b/docs/how-to/metric-snapshot-s3-bench.md @@ -0,0 +1,79 @@ +# Metric snapshot benchmark on AWS S3 + +The manual `Metric Snapshot S3 Benchmark` workflow tests packed Metric export and +restore against real S3 in `ap-southeast-1`. It must first exist on the repository's +default branch for GitHub to register `workflow_dispatch`. It was validated on AWS +by run [38544464731](https://github.com/GreptimeTeam/greptimedb/actions/runs/38544464731); +[#9573](https://github.com/GreptimeTeam/greptimedb/pull/9573) records its conditions and results. + +## Run + +Use the repository's existing EC2 runner configuration and CI S3 credentials. +The workflow passes credentials only through CI; no local AWS credentials are +needed. Source, target and CLI use the same immutable image. This image's source +SHA and required CLI capabilities have been checked: + +```bash +gh workflow run metric-snapshot-s3-bench.yml \ + --repo GreptimeTeam/greptimedb --ref main \ + -f greptimedb_image=greptime/greptimedb-nightly@sha256:20a1182bed0f452e6d573ca110b313e0b5cb76de7aedfa41a0c339963d4683d3 \ + -f expected_commit=f690bbb2040865c0d22b0b57d51612b60ae27cd0 \ + -f logical_tables_per_group=10000 \ + -f benchmark_timeout_minutes=240 \ + -f prior_spend_usd=0 +``` + +Start a new dispatch for each attempt and update `prior_spend_usd` to include all +earlier cloud attempts in the campaign, including failed setup and cleanup. +Do not reset the estimate to zero when retrying. The workflow accepts 50–10,000 +logical tables per group and a 30–240 minute benchmark job timeout. + +The runner is fixed at `c6id.4xlarge` (16 vCPU, 32 GiB RAM). At the recorded +Singapore Linux on-demand estimate of $0.9408/hour, the default attempt reserves +$6.4392: four benchmark hours, 45 minutes of provisioning, 10 minutes for queueing, +20 minutes of cleanup, and $1.50 for S3, disk and IPv4 costs. Preflight rejects prior spend at or above +$15, or an attempt whose reserve would exceed the cumulative $20 budget. +These estimates and checks are not an AWS hard spending cap or an AWS bill. + +## Workload and validation + +- A 10-physical-table × 50-logical-table smoke runs first. Its 500 logical tables + cover all 298 string tag columns before the larger workload starts. +- The default full run uses 10 physical tables and 100,000 logical tables, with + 32 rows per logical table: 3.2 million rows. Each logical schema has 10 + overlapping tags, a timestamp and a DOUBLE; schemas vary between tables. +- Source and target use local File storage on the temporary NVMe disk. Only + snapshots use S3. Source validation finishes before the source process stops; + restoration then starts a separate target process. +- Export uses `--pack-metric-data`, parallelism 4, chunk parallelism 1, and one + complete time chunk. Import uses task parallelism 1 and automatic batch DDL. +- Every source and restored logical table is checked for membership, schema, + DDL and typed values including NULLs. Target placeholders force table ID + remapping; successful batch DDL and removal of import resume state are checked. + +## Results and cleanup + +The job summary reports complete export, structural verify and import durations; +data generation and full data validation are timed separately. Evidence artifacts +are credential-scanned and retained for 14 days. Keep conclusions limited to +this version's absolute AWS results; this workflow does not compare old versions. + +Artifacts include sampled CPU/RSS, host memory, swap/OOM counters, server metrics, +CLI logs and S3 object/byte inventories. Configured swap is not RAM. A phase with +new swap-out or OOM fails and retains its evidence. OpenDAL operation counters +are not complete AWS HTTP request counts: multipart calls, retries and CLI traffic +are not fully covered. Object inventory bytes are not transferred bytes. + +A separate GitHub-hosted teardown job starts when provisioning finishes and polls +the benchmark job. Its deadline includes queue time: allocation start plus 45 +minutes, the benchmark timeout and 10 minutes. Completion, cancellation, an API +error or the deadline leads to cleanup. Failed provisioning skips the wait. +It terminates the EC2 instance and its disks, unregisters the runner, deletes every +object listed under the run's unique prefix and verifies the listing is empty. Like +the object store WAL benchmark, it needs only list, put, get and delete on the CI +bucket. It does not abort incomplete multipart uploads or remove noncurrent object +versions; those depend on the bucket's lifecycle configuration. Check teardown completion before another +attempt. A deadline may leave the GitHub benchmark job queued after its resources +are removed; cancel that workflow manually. If cleanup fails, rerun the teardown +job only; it keeps the original attempt's resource identity and deadline. A new benchmark needs a new dispatch because teardown +removes its runner. The daily EC2 janitor is only a fallback and does not clean S3.