mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-14 23:06:59 +00:00
test: add manual S3 metric snapshot benchmark (#9573)
* test: add repeatable metric snapshot S3 benchmark
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* 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 <fengjiachun@gmail.com>
* 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 <fengjiachun@gmail.com>
* docs: record the AWS validation run of the metric snapshot benchmark
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
---------
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
(cherry picked from commit 7df99f946e)
This commit is contained in:
jeremyhi
discord9
@@ -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()
|
||||
@@ -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/<run>-<attempt>/")
|
||||
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()
|
||||
@@ -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/<run_id>-<attempt>/")
|
||||
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 <run prefix><round>/; 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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user