From 4b527430c1c28b15bc06c4bf8084d430d3b42133 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 13:24:11 -0600 Subject: [PATCH 1/2] feat(o11y): compare matched exact results and scoped runtime measurements --- tools/o11y-execution/README.md | 44 +++++++++- tools/o11y-execution/compare.py | 126 +++++++++++++++++++++++++++ tools/o11y-execution/replay.py | 64 +++++++++++++- tools/o11y-execution/test_compare.py | 78 +++++++++++++++++ 4 files changed, 307 insertions(+), 5 deletions(-) create mode 100644 tools/o11y-execution/compare.py create mode 100644 tools/o11y-execution/test_compare.py diff --git a/tools/o11y-execution/README.md b/tools/o11y-execution/README.md index 6d9f9226..b9411f8f 100644 --- a/tools/o11y-execution/README.md +++ b/tools/o11y-execution/README.md @@ -60,5 +60,45 @@ records accepted batches; acceptance does not prove worker completion. The settle interval is recorded, not a completion barrier. The first traversal is called `first_pass`, not “cold cache”; later traversals are `repeat`. All failures and fallback responses remain in the denominator. A completion file means the -replay finished, **not** that accuracy or benefit acceptance passed. The next -stacked PR adds matched exact queries and measurement/reporting. +replay finished, **not** that accuracy or benefit acceptance passed. + +## Matched exact comparison + +Add `--compare --exact-pid PID` to the same command. PID must identify the local +Prometheus serving `--exact-url`; verify that association before running. If it is +remote, omit PID: exact-service CPU and memory stay unavailable. The runner never +stops this externally managed process. Prometheus must have the same data and +evaluation range; both endpoints receive identical encoded sample batches. + +Every occurrence is queried against both endpoints with identical PromQL and +time, alternating request order. Results are matched by full label set and sample +timestamp, not row order. `comparison.json` reports missing/extra series and +samples, completeness, absolute/relative error, zero-denominator mismatches, +failures, first-pass/repeat latency distributions and sequential service rate. +Duplicate series, failed responses, unsupported response types and warnings make +a result uncomparable. Matching one dataset is not a formal confidence guarantee. + +The latency ratio is emitted only when **all** matched responses are exact-equal +and successful. Approximate discrepancies are still reported, but no arbitrary +error threshold is substituted for each query's accuracy contract. Fallbacks +remain in the totals. The ratio measures query service time only, never amortized +end-to-end savings. Raw per-request timings allow other analyses without hiding +the unsuccessful portion of the workload. + +Linux process counters are captured around HTTP calls and at startup, ingestion +and query boundaries. Backend calls also record exact-service CPU when available, +so forwarded work is visible. Ingestion journal timings and phase snapshots expose +the construction/update interval, including the declared settle delay. Counters +include background work and have CPU-tick resolution. RSS/HWM are **whole-process** +memory, not summary heap; HWM is process-lifetime peak, not isolated phase peak. +`store.json` preserves the backend's raw state counters. Affinity/cgroup metadata +is recorded but equal resource budgets are not enforced by this harness. + +The provider's estimated costs are preserved next to measured quantities without +pretending abstract model units are CPU nanoseconds. End-to-end benefit and +estimated/measured cost ratios remain null until their units, lifecycle scope, +exact-service startup/storage costs and resource budgets are matched. Separate +fresh-process/cache-controlled trials, a retained-state measurement, calibration +provenance and real-corpus execution evidence are still required before declaring +the five #524 acceptance criteria complete. The shared fallback/baseline service +can transfer cache warmth; alternating order does not eliminate this confound. diff --git a/tools/o11y-execution/compare.py b/tools/o11y-execution/compare.py new file mode 100644 index 00000000..7171eb10 --- /dev/null +++ b/tools/o11y-execution/compare.py @@ -0,0 +1,126 @@ +"""Matched Prometheus-result comparison and explicitly scoped process measurements.""" +import math +import os +from pathlib import Path +import statistics + + +def result_samples(response): + if response.get("status") != "success": + raise ValueError("query did not succeed") + if response.get("warnings"): + raise ValueError("response warnings require review for partial results") + data = response["data"] + kind = data["resultType"] + items = data["result"] + if kind == "scalar": + items = [{"metric": {}, "value": items}] + elif kind not in ("vector", "matrix"): + raise ValueError(f"unsupported comparison result type: {kind}") + result, groups = {}, set() + for item in items: + labels = tuple(sorted(item["metric"].items())) + if labels in groups: + raise ValueError("duplicate series") + groups.add(labels) + for timestamp, value in item["values"] if kind == "matrix" else [item["value"]]: + key = (labels, float(timestamp)) + if key in result or not math.isfinite(key[1]): + raise ValueError("duplicate or invalid sample timestamp") + result[key] = float(value) + return kind, groups, result + + +def compare_results(actual, expected): + try: + ak, ag, a = result_samples(actual) + ek, eg, e = result_samples(expected) + if ak != ek: + raise ValueError("different result types") + except (ValueError, TypeError, KeyError) as error: + return {"comparable": False, "equal": False, "reason": str(error)} + absolute, relative, zero, nonfinite = [], [], 0, 0 + for key in a.keys() & e.keys(): + x, y = a[key], e[key] + if not math.isfinite(x) or not math.isfinite(y): + if not (x == y or (math.isnan(x) and math.isnan(y))): + nonfinite += 1 + continue + error = abs(x - y) + if not math.isfinite(error): + nonfinite += 1 + continue + absolute.append(error) + if y: + ratio = error / abs(y) + if math.isfinite(ratio): + relative.append(ratio) + else: + nonfinite += 1 + elif error: + zero += 1 + missing, extra = len(e.keys() - a.keys()), len(a.keys() - e.keys()) + return {"comparable": True, + "equal": ag == eg and not (missing or extra or zero or nonfinite or any(absolute)), + "missing_series": len(eg - ag), "extra_series": len(ag - eg), + "missing_samples": missing, "extra_samples": extra, + "completeness": (len(a.keys() & e.keys()) / len(e)) if e else (1.0 if not a else 0.0), + "max_absolute_error": max(absolute, default=None), + "max_relative_error": max(relative, default=None), + "zero_baseline_mismatches": zero, "nonfinite_mismatches": nonfinite} + + +def distribution(values): + if not values: + return None + values = sorted(values) + return {"count": len(values), "min_ns": values[0], "median_ns": statistics.median(values), + "p95_ns": values[math.ceil(0.95 * len(values)) - 1], "max_ns": values[-1], + "mean_ns": statistics.mean(values), "stddev_ns": statistics.stdev(values) if len(values) > 1 else None} + + +def summarize(rows): + comparisons = [compare_results(row["response"], row.get("exact", {}).get("response", {})) for row in rows] + eligible = bool(rows) and all( + c["equal"] and r["execution"] in ("warm", "exact_fallback") and r["exact"].get("http_status") == 200 + for r, c in zip(rows, comparisons)) + actual = sum(r["elapsed_ns"] for r in rows) + exact = sum(r.get("exact", {}).get("elapsed_ns", 0) for r in rows) + return {"occurrences": len(rows), + "execution_counts": {k: sum(r["execution"] == k for r in rows) for k in ("warm", "exact_fallback", "failed")}, + "equal_results": sum(c["equal"] for c in comparisons), + "uncomparable_results": sum(not c["comparable"] for c in comparisons), + "comparisons": comparisons, + "backend_latency": distribution([r["elapsed_ns"] for r in rows]), + "exact_latency": distribution([r["exact"]["elapsed_ns"] for r in rows if "exact" in r]), + "successful_backend_requests_per_query_service_second": + sum(r["execution"] != "failed" for r in rows) * 1e9 / actual if actual else None, + "matched_query_latency_ratio": exact / actual if eligible and actual else None, + "end_to_end_benefit": None, + "scope": "sequential HTTP service time including failures; not concurrent/system throughput; ratio only for wholly exact-equal matched results"} + + +def process_snapshot(pid): + """Linux process scope, not retained summary heap. PID reuse invalidates deltas.""" + try: + root = Path(f"/proc/{pid}") + stat = (root / "stat").read_text().rsplit(")", 1)[1].split() + status = dict(line.split(":", 1) for line in (root / "status").read_text().splitlines() if ":" in line) + return {"pid": pid, "start_ticks": int(stat[19]), + "cpu_ns": int((int(stat[11]) + int(stat[12])) * 1e9 / os.sysconf("SC_CLK_TCK")), + "rss_bytes": int(status["VmRSS"].split()[0]) * 1024, + "process_lifetime_peak_rss_bytes": int(status["VmHWM"].split()[0]) * 1024, + "cpu_affinity": status.get("Cpus_allowed_list", "").strip(), + "cgroup": (root / "cgroup").read_text(), + "retained_summary_state_bytes": None} + except (OSError, KeyError, ValueError, IndexError): + return None + + +def process_delta(before, after): + if not before or not after or (before["pid"], before["start_ticks"]) != (after["pid"], after["start_ticks"]): + return None + return {"cpu_ns": after["cpu_ns"] - before["cpu_ns"], + "rss_before_bytes": before["rss_bytes"], "rss_after_bytes": after["rss_bytes"], + "process_lifetime_peak_rss_bytes": after["process_lifetime_peak_rss_bytes"], + "scope": "whole process including background work; CPU tick resolution; HWM is lifetime, not phase peak"} diff --git a/tools/o11y-execution/replay.py b/tools/o11y-execution/replay.py index 942ab506..5f8ad8ea 100644 --- a/tools/o11y-execution/replay.py +++ b/tools/o11y-execution/replay.py @@ -15,6 +15,10 @@ import urllib.parse import urllib.request +from compare import compare_results, process_snapshot, process_delta, summarize + +PROCESS_IDS = {} + def classify(response, headers=None): if response.get("status") != "success": @@ -124,7 +128,7 @@ def encode_write(rows): return varint(length) + literal + wire -def request(url, data=None, headers=None): +def _http_request(url, data=None, headers=None): start = time.perf_counter_ns() try: with urllib.request.urlopen(urllib.request.Request(url, data=data, headers=headers or {}), timeout=60) as response: @@ -140,6 +144,18 @@ def request(url, data=None, headers=None): "headers": {}, "elapsed_ns": time.perf_counter_ns() - start} +def process_snapshots(): + return {name: process_snapshot(pid) for name, pid in PROCESS_IDS.items()} + + +def request(url, data=None, headers=None): + before = process_snapshots() + result = _http_request(url, data, headers) + after = process_snapshots() + result["process_resources"] = {name: process_delta(before.get(name), after.get(name)) for name in PROCESS_IDS} + return result + + def write_json(path, value): path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n") @@ -164,17 +180,27 @@ def ingest(rows, endpoints, output): raise RuntimeError(f"ingestion failed at batch {offset}; inspect partial acceptance before retry") -def replay(queries, backend, output, repetitions): +def replay(queries, backend, output, repetitions, exact_url=None): rows = [] for repeat in range(repetitions): for query in queries: params = urllib.parse.urlencode({"query": query["query"], "time": f'{query["eval_timestamp_ms"] / 1000:.3f}'}) + # Alternate paired order to expose, rather than always favor, cache/order effects. + exact = None + if exact_url and (repeat + len(rows)) % 2 == 0: + exact = request(exact_url.rstrip("/") + "/api/v1/query?" + params) answer = request(backend.rstrip("/") + "/api/v1/query?" + params) + if exact_url and exact is None: + exact = request(exact_url.rstrip("/") + "/api/v1/query?" + params) route = classify(answer["response"], answer["headers"]) if answer["http_status"] != 200: route = "failed" rows.append({**query, "repetition": repeat, "phase": "first_pass" if repeat == 0 else "repeat", "execution": route, **answer}) + if exact is not None: + rows[-1]["exact"] = exact + rows[-1]["comparison"] = compare_results(answer["response"], exact["response"]) + rows[-1]["pair_order"] = "exact_first" if (repeat + len(rows) - 1) % 2 == 0 else "backend_first" write_json(output / "queries.json", rows) return rows @@ -187,6 +213,8 @@ def main(): parser.add_argument("--compiler", type=Path, required=True) parser.add_argument("--data-plane", type=Path, required=True) parser.add_argument("--exact-url", required=True, help="dedicated empty Prometheus with Remote Write receiver enabled") + parser.add_argument("--compare", action="store_true", help="execute a matched exact request for every corpus occurrence") + parser.add_argument("--exact-pid", type=int, help="local Prometheus PID for Linux CPU/RSS evidence; never stopped by this runner") parser.add_argument("--port", type=int, default=18089) parser.add_argument("--settle-seconds", type=float, default=2) parser.add_argument("--repetitions", type=int, default=2) @@ -197,6 +225,8 @@ def main(): corpus = json.loads(args.queries.read_text()) queries = validate_workload(json.loads(args.snapshot.read_text()), corpus) samples = parse_samples(args.metrics.read_text().splitlines()) + if args.exact_pid is not None and process_snapshot(args.exact_pid) is None: + parser.error("exact-pid must name a readable live local process") with socket.socket() as probe: probe.bind(("127.0.0.1", args.port)) args.output.mkdir(parents=True, exist_ok=False) @@ -222,6 +252,10 @@ def main(): write_json(args.output / "command.json", command) with (args.output / "backend.log").open("w") as log: child = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT) + PROCESS_IDS["backend"] = child.pid + if args.exact_pid is not None: + PROCESS_IDS["exact_service"] = args.exact_pid + phases = {"startup": process_snapshots()} try: for _ in range(120): if child.poll() is not None: @@ -239,13 +273,37 @@ def main(): if not any(p["plan_id"] == envelope["plan_id"] and p["plan_version"] == envelope["plan_version"] and p["phase"] == "active" for p in installed["response"].get("plans", [])): raise RuntimeError("runtime has not activated the selected plan generation") + phases["before_ingest"] = process_snapshots() + ingest_start = time.perf_counter_ns() ingest(samples, [args.exact_url, backend], args.output) time.sleep(args.settle_seconds) - results = replay(queries, backend, args.output, args.repetitions) + ingest_elapsed = time.perf_counter_ns() - ingest_start + phases["after_ingest_and_settle"] = process_snapshots() + write_json(args.output / "process-phases.json", phases) + results = replay(queries, backend, args.output, args.repetitions, args.exact_url if args.compare else None) + phases["after_queries"] = process_snapshots() + write_json(args.output / "process-phases.json", phases) write_json(args.output / "store.json", request(backend + "/api/v1/store/metrics")) write_json(args.output / "completion.json", {"complete": True, "execution_counts": {k: sum(r["execution"] == k for r in results) for k in ["warm", "exact_fallback", "failed"]}, "benefit_claim": None}) + if args.compare: + report = {"schema_version": 1, "all_requests": summarize(results), + "by_phase": {phase: summarize([r for r in results if r["phase"] == phase]) + for phase in ["first_pass", "repeat"]}, + "estimated_cost": plan["cost_comparison"], + "measured_ingest_and_settle_wall_ns": ingest_elapsed, + "planning_wall_ns": plan["planning_elapsed_ns"], + "process_phases": phases, + "estimated_vs_measured_cost_ratio": None, + "acceptance_complete": False, + "limitations": ["No common conversion from provider cost units to measured resource units", + "Exact service is shared with fallback; paired order alternates but caches are not isolated", + "Resource budgets must be independently matched; affinity/cgroup recorded, not enforced", + "Raw process RSS is not summary state size; store.json retains backend counters", + "Planning CPU, exact-service construction/storage and isolated cold-start runs remain unmeasured", + "Query equality on one dataset is not a formal approximation confidence guarantee"]} + write_json(args.output / "comparison.json", report) finally: child.terminate() try: diff --git a/tools/o11y-execution/test_compare.py b/tools/o11y-execution/test_compare.py new file mode 100644 index 00000000..800b66a8 --- /dev/null +++ b/tools/o11y-execution/test_compare.py @@ -0,0 +1,78 @@ +"""Correctness gates must survive missing rows, failed responses and zero baselines.""" +import unittest +import tempfile +from pathlib import Path +from unittest.mock import patch +from urllib.parse import urlparse, parse_qs +from compare import compare_results, summarize, process_delta + + +def vector(*values): + return {"status": "success", "data": {"resultType": "vector", "result": [ + {"metric": {"job": key}, "value": [123, str(value)]} for key, value in values]}} + + +class ComparisonTests(unittest.TestCase): + def test_group_matching_is_not_row_position(self): + """Permuting a group-by result does not change correctness.""" + result = compare_results(vector(("b", 2), ("a", 1)), vector(("a", 1), ("b", 2))) + self.assertTrue(result["equal"]) + self.assertEqual(result["completeness"], 1) + + def test_missing_rows_and_zero_baseline_cannot_hide_errors(self): + """Report structural loss separately; zero denominators are not epsilon-clamped.""" + result = compare_results(vector(("a", 2)), vector(("a", 0), ("b", 3))) + self.assertFalse(result["equal"]) + self.assertEqual(result["missing_series"], 1) + self.assertEqual(result["max_absolute_error"], 2) + self.assertEqual(result["zero_baseline_mismatches"], 1) + self.assertIsNone(result["max_relative_error"]) + + def test_duplicates_and_errors_are_not_equal(self): + """Malformed or failed responses cannot become a successful empty comparison.""" + for response in [{"status": "error"}, vector(("a", 1), ("a", 1))]: + self.assertFalse(compare_results(response, vector(("a", 1)))["comparable"]) + + def test_failed_or_missing_baseline_prevents_benefit_claim(self): + """Failures remain in the workload denominator; no successful-subset speedup.""" + row = {"execution": "warm", "phase": "repeat", "elapsed_ns": 10, + "response": vector(("a", 1)), "exact": {"http_status": 500, + "elapsed_ns": 100, "response": {"status": "error"}}} + report = summarize([row]) + self.assertEqual(report["occurrences"], 1) + self.assertIsNone(report["matched_query_latency_ratio"]) + self.assertIsNone(report["end_to_end_benefit"]) + + def test_matrix_requires_the_same_timestamps_and_empty_series(self): + """Series and time coverage are part of correctness even without numeric samples.""" + empty = {"status": "success", "data": {"resultType": "matrix", "result": []}} + series = {"status": "success", "data": {"resultType": "matrix", "result": [ + {"metric": {"job": "a"}, "values": []}]}} + self.assertFalse(compare_results(empty, series)["equal"]) + + def test_pid_reuse_and_unreadable_processes_are_not_free_cpu(self): + """A missing counter or a different process lifetime must remain unknown.""" + self.assertIsNone(process_delta(None, None)) + self.assertIsNone(process_delta({"pid": 1, "start_ticks": 1}, {"pid": 1, "start_ticks": 2})) + + def test_paired_replay_preserves_occurrences_time_and_alternates_order(self): + """Harness conformance only: stub responses are never benchmark evidence.""" + from replay import replay + calls = [] + def respond(url): + calls.append(url) + response = vector(("a", 1)) + response["infos"] = ["data_source: asap_query"] + return {"response": response, "http_status": 200, "headers": {}, "elapsed_ns": 10} + queries = [{"id": x, "query": "sum(up{job=\"a\"})", "eval_timestamp_ms": 1234567} for x in ["a", "b"]] + with tempfile.TemporaryDirectory() as folder, patch("replay.request", side_effect=respond): + rows = replay(queries, "http://backend", Path(folder), 2, "http://exact") + self.assertEqual(len(rows), 4) + self.assertTrue(all(r["comparison"]["equal"] for r in rows)) + self.assertEqual([urlparse(u).netloc for u in calls[:4]], ["exact", "backend", "backend", "exact"]) + for url in calls: + self.assertEqual(parse_qs(urlparse(url).query), {"query": [queries[0]["query"]], "time": ["1234.567"]}) + + +if __name__ == "__main__": + unittest.main() From b0fe779b3d55b982a462191361b2d77af0867bb0 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 14:24:32 -0600 Subject: [PATCH 2/2] fix: isolate comparison fallback and account for query resources --- tools/o11y-execution/README.md | 17 +++- tools/o11y-execution/compare.py | 13 ++- tools/o11y-execution/replay.py | 114 ++++++++++++++++++++++++--- tools/o11y-execution/test_compare.py | 14 ++++ tools/o11y-execution/test_replay.py | 15 +++- 5 files changed, 157 insertions(+), 16 deletions(-) diff --git a/tools/o11y-execution/README.md b/tools/o11y-execution/README.md index b9411f8f..f2237ce7 100644 --- a/tools/o11y-execution/README.md +++ b/tools/o11y-execution/README.md @@ -91,8 +91,19 @@ so forwarded work is visible. Ingestion journal timings and phase snapshots expo the construction/update interval, including the declared settle delay. Counters include background work and have CPU-tick resolution. RSS/HWM are **whole-process** memory, not summary heap; HWM is process-lifetime peak, not isolated phase peak. -`store.json` preserves the backend's raw state counters. Affinity/cgroup metadata -is recorded but equal resource budgets are not enforced by this harness. +`store.json` preserves the backend's raw state counters. `--cpu-affinity` applies +one CPU set to the backend and supplied Prometheus processes/threads. +`--address-space-bytes` applies the same RLIMIT_AS, which limits virtual address +space, not RSS or combined process memory. These controls are optional and their +presence is recorded; they do not establish a full cgroup resource budget. + +Use `--fallback-url` and `--fallback-pid` for a separate fresh Prometheus instance, +keeping `--exact-url`/`--exact-pid` for the baseline. Identical process IDs and +storage paths are rejected. Both receive identical input batches. Backend CPU +includes its fallback service CPU; baseline CPU remains separate. With no +separate fallback service, the report preserves the shared-cache limitation. +`--exact-storage` and `--fallback-storage` record logical file sizes separately; +backend output file sizes include logs and are not retained summary heap sizes. The provider's estimated costs are preserved next to measured quantities without pretending abstract model units are CPU nanoseconds. End-to-end benefit and @@ -100,5 +111,5 @@ estimated/measured cost ratios remain null until their units, lifecycle scope, exact-service startup/storage costs and resource budgets are matched. Separate fresh-process/cache-controlled trials, a retained-state measurement, calibration provenance and real-corpus execution evidence are still required before declaring -the five #524 acceptance criteria complete. The shared fallback/baseline service +the five #524 acceptance criteria complete. A shared fallback/baseline service can transfer cache warmth; alternating order does not eliminate this confound. diff --git a/tools/o11y-execution/compare.py b/tools/o11y-execution/compare.py index 7171eb10..09d7da22 100644 --- a/tools/o11y-execution/compare.py +++ b/tools/o11y-execution/compare.py @@ -86,7 +86,18 @@ def summarize(rows): for r, c in zip(rows, comparisons)) actual = sum(r["elapsed_ns"] for r in rows) exact = sum(r.get("exact", {}).get("elapsed_ns", 0) for r in rows) - return {"occurrences": len(rows), + def cpu_total(requests, names): + values = [request.get("process_resources", {}).get(name) for request in requests for name in names] + return sum(v["cpu_ns"] for v in values) if values and all(v is not None for v in values) else None + has_fallback_process = any("fallback_service" in r.get("process_resources", {}) for r in rows) + backend_names = ["backend", "fallback_service" if has_fallback_process else "exact_service"] + backend_cpu = cpu_total(rows, backend_names) + exact_cpu = cpu_total([r.get("exact", {}) for r in rows], ["exact_service"]) + return {"backend_plus_fallback_cpu_ns": backend_cpu, + "baseline_cpu_ns": exact_cpu, + "baseline_over_backend_cpu_ratio": exact_cpu / backend_cpu if eligible and backend_cpu and exact_cpu is not None else None, + "cpu_scope": "request intervals, whole processes including background work; fallback CPU charged to backend; /proc tick granularity", + "occurrences": len(rows), "execution_counts": {k: sum(r["execution"] == k for r in rows) for k in ("warm", "exact_fallback", "failed")}, "equal_results": sum(c["equal"] for c in comparisons), "uncomparable_results": sum(not c["comparable"] for c in comparisons), diff --git a/tools/o11y-execution/replay.py b/tools/o11y-execution/replay.py index 5f8ad8ea..405b7541 100644 --- a/tools/o11y-execution/replay.py +++ b/tools/o11y-execution/replay.py @@ -5,6 +5,8 @@ import hashlib import json import math +import os +import resource from pathlib import Path import re import socket @@ -20,6 +22,18 @@ PROCESS_IDS = {} +def constrain_process(pid, cpus, address_space_bytes=None): + """Match schedulable CPUs for all existing threads, including Go workers.""" + if cpus: + for task in Path(f"/proc/{pid}/task").iterdir(): + try: + os.sched_setaffinity(int(task.name), cpus) + except ProcessLookupError: + pass + if address_space_bytes: + resource.prlimit(pid, resource.RLIMIT_AS, (address_space_bytes, address_space_bytes)) + + def classify(response, headers=None): if response.get("status") != "success": return "failed" @@ -139,6 +153,15 @@ def _http_request(url, data=None, headers=None): body = {"raw": body.decode(errors="replace")} return {"http_status": status, "response": body, "headers": {k.lower(): v for k, v in received.items()}, "elapsed_ns": time.perf_counter_ns() - start} + except urllib.error.HTTPError as error: + raw = error.read() + try: + body = json.loads(raw) + except (ValueError, UnicodeDecodeError): + body = {"status": "error", "error": str(error), "raw": raw.decode(errors="replace")} + return {"http_status": error.code, "response": body, + "headers": {k.lower(): v for k, v in error.headers.items()}, + "elapsed_ns": time.perf_counter_ns() - start} except (OSError, urllib.error.URLError) as error: return {"http_status": getattr(error, "code", None), "response": {"status": "error", "error": str(error)}, "headers": {}, "elapsed_ns": time.perf_counter_ns() - start} @@ -183,11 +206,12 @@ def ingest(rows, endpoints, output): def replay(queries, backend, output, repetitions, exact_url=None): rows = [] for repeat in range(repetitions): - for query in queries: + for query_index, query in enumerate(queries): + exact_first = (repeat + query_index) % 2 == 0 params = urllib.parse.urlencode({"query": query["query"], "time": f'{query["eval_timestamp_ms"] / 1000:.3f}'}) # Alternate paired order to expose, rather than always favor, cache/order effects. exact = None - if exact_url and (repeat + len(rows)) % 2 == 0: + if exact_url and exact_first: exact = request(exact_url.rstrip("/") + "/api/v1/query?" + params) answer = request(backend.rstrip("/") + "/api/v1/query?" + params) if exact_url and exact is None: @@ -200,7 +224,7 @@ def replay(queries, backend, output, repetitions, exact_url=None): if exact is not None: rows[-1]["exact"] = exact rows[-1]["comparison"] = compare_results(answer["response"], exact["response"]) - rows[-1]["pair_order"] = "exact_first" if (repeat + len(rows) - 1) % 2 == 0 else "backend_first" + rows[-1]["pair_order"] = "exact_first" if exact_first else "backend_first" write_json(output / "queries.json", rows) return rows @@ -215,6 +239,12 @@ def main(): parser.add_argument("--exact-url", required=True, help="dedicated empty Prometheus with Remote Write receiver enabled") parser.add_argument("--compare", action="store_true", help="execute a matched exact request for every corpus occurrence") parser.add_argument("--exact-pid", type=int, help="local Prometheus PID for Linux CPU/RSS evidence; never stopped by this runner") + parser.add_argument("--exact-storage", type=Path, help="baseline Prometheus data directory for logical on-disk byte count") + parser.add_argument("--fallback-storage", type=Path, help="fallback Prometheus data directory for logical on-disk byte count") + parser.add_argument("--fallback-url", help="separate fresh Prometheus for backend fallback; defaults to exact-url") + parser.add_argument("--fallback-pid", type=int) + parser.add_argument("--cpu-affinity", help="comma-separated permitted CPU IDs; enforced on backend and supplied Prometheus PIDs") + parser.add_argument("--address-space-bytes", type=int, help="same RLIMIT_AS for backend and supplied Prometheus; virtual memory, not RSS cap") parser.add_argument("--port", type=int, default=18089) parser.add_argument("--settle-seconds", type=float, default=2) parser.add_argument("--repetitions", type=int, default=2) @@ -222,6 +252,27 @@ def main(): args = parser.parse_args() if args.repetitions < 1 or not 0 <= args.settle_seconds <= 60: parser.error("positive repetitions and settle-seconds in [0, 60] required") + cpus = {int(x) for x in args.cpu_affinity.split(",")} if args.cpu_affinity else None + if cpus and not cpus <= os.sched_getaffinity(0): + parser.error("requested CPUs must be available to the runner") + if args.address_space_bytes is not None and args.address_space_bytes <= 0: + parser.error("address-space-bytes must be positive") + if (cpus or args.address_space_bytes) and not args.exact_pid: + parser.error("enforced resource comparison requires --exact-pid") + if args.fallback_url and args.fallback_url.rstrip("/") == args.exact_url.rstrip("/"): + parser.error("fallback-url must be a separate service") + if args.fallback_url and (cpus or args.address_space_bytes) and not args.fallback_pid: + parser.error("resource enforcement also requires --fallback-pid") + if args.fallback_url and args.exact_pid is not None and args.exact_pid == args.fallback_pid: + parser.error("baseline and fallback must use distinct processes") + if args.exact_storage and args.fallback_storage and args.exact_storage.resolve() == args.fallback_storage.resolve(): + parser.error("baseline and fallback must use distinct storage directories") + fallback_url = args.fallback_url or args.exact_url + for pid in [args.exact_pid, args.fallback_pid]: + if pid is not None: + if process_snapshot(pid) is None: + parser.error("service PID must be readable and live") + constrain_process(pid, cpus, args.address_space_bytes) corpus = json.loads(args.queries.read_text()) queries = validate_workload(json.loads(args.snapshot.read_text()), corpus) samples = parse_samples(args.metrics.read_text().splitlines()) @@ -235,26 +286,38 @@ def main(): for p in [args.metrics, args.queries, args.snapshot, args.compiler, args.data_plane]}, "samples": len(samples), "timestamp_min_ms": samples[0][2], "timestamp_max_ms": samples[-1][2], "query_occurrences": len(queries), "configuration": {k: str(v) for k, v in vars(args).items()}, - "limitations": ["generator provenance must be supplied separately", "first pass is not a guaranteed cold cache", "ingest acceptance is not materialization completion"]} + "limitations": ["generator provenance must be supplied separately", "first pass is not a guaranteed cold cache", "ingest acceptance and settle time do not prove materialization completion"]} write_json(args.output / "run.json", provenance) + planning_before = resource.getrusage(resource.RUSAGE_CHILDREN) with (args.output / "planning.stderr").open("w") as log: compiled = subprocess.run([str(args.compiler.resolve()), str(args.snapshot.resolve())], check=True, stdout=subprocess.PIPE, stderr=log, text=True) + planning_after = resource.getrusage(resource.RUSAGE_CHILDREN) + planning_resources = {"cpu_ns": int(((planning_after.ru_utime + planning_after.ru_stime) - + (planning_before.ru_utime + planning_before.ru_stime)) * 1e9), + "children_lifetime_peak_rss_bytes": planning_after.ru_maxrss * 1024} plan = json.loads(compiled.stdout) write_json(args.output / "planning.json", plan) artifact = args.output / "install.json" write_json(artifact, plan["install_request"]) backend = f"http://127.0.0.1:{args.port}" command = [str(args.data_plane.resolve()), "--profile", "asapquery", "--physical-plan", str(artifact.resolve()), - "--prometheus-server", args.exact_url, "--forward-unsupported-queries", "--http-port", str(args.port), + "--prometheus-server", fallback_url, "--forward-unsupported-queries", "--http-port", str(args.port), "--output-dir", str((args.output / "backend").resolve()), "--precompute-allowed-lateness-ms", "0", "--precompute-flush-interval-ms", "25"] write_json(args.output / "command.json", command) with (args.output / "backend.log").open("w") as log: - child = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT) + def limits(): + if cpus: + os.sched_setaffinity(0, cpus) + if args.address_space_bytes: + resource.setrlimit(resource.RLIMIT_AS, (args.address_space_bytes, args.address_space_bytes)) + child = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT, preexec_fn=limits) PROCESS_IDS["backend"] = child.pid if args.exact_pid is not None: PROCESS_IDS["exact_service"] = args.exact_pid + if args.fallback_pid is not None: + PROCESS_IDS["fallback_service"] = args.fallback_pid phases = {"startup": process_snapshots()} try: for _ in range(120): @@ -275,15 +338,25 @@ def main(): raise RuntimeError("runtime has not activated the selected plan generation") phases["before_ingest"] = process_snapshots() ingest_start = time.perf_counter_ns() - ingest(samples, [args.exact_url, backend], args.output) + ingest(samples, list(dict.fromkeys([args.exact_url, fallback_url, backend])), args.output) time.sleep(args.settle_seconds) ingest_elapsed = time.perf_counter_ns() - ingest_start phases["after_ingest_and_settle"] = process_snapshots() + write_json(args.output / "store-after-build.json", request(backend + "/api/v1/store/metrics")) write_json(args.output / "process-phases.json", phases) results = replay(queries, backend, args.output, args.repetitions, args.exact_url if args.compare else None) phases["after_queries"] = process_snapshots() write_json(args.output / "process-phases.json", phases) - write_json(args.output / "store.json", request(backend + "/api/v1/store/metrics")) + store = request(backend + "/api/v1/store/metrics") + write_json(args.output / "store.json", store) + def disk_bytes(path): + return sum(p.stat().st_size for p in path.rglob("*") if p.is_file()) if path else None + storage = {"backend_output_bytes": disk_bytes(args.output / "backend"), + "baseline_prometheus_bytes": disk_bytes(args.exact_storage), + "fallback_prometheus_bytes": disk_bytes(args.fallback_storage), + "backend_store": store, + "scope": "logical file bytes including WAL; backend output also contains logs; concurrent snapshots are approximate"} + write_json(args.output / "storage.json", storage) write_json(args.output / "completion.json", {"complete": True, "execution_counts": {k: sum(r["execution"] == k for r in results) for k in ["warm", "exact_fallback", "failed"]}, "benefit_claim": None}) @@ -291,17 +364,36 @@ def main(): report = {"schema_version": 1, "all_requests": summarize(results), "by_phase": {phase: summarize([r for r in results if r["phase"] == phase]) for phase in ["first_pass", "repeat"]}, + "by_query_occurrence": {query["id"]: summarize([r for r in results if r["id"] == query["id"]]) + for query in queries}, + "by_execution": {route: summarize([r for r in results if r["execution"] == route]) + for route in ["warm", "exact_fallback", "failed"]}, "estimated_cost": plan["cost_comparison"], + "measurement_units": {"latency": "nanoseconds", "cpu": "process CPU nanoseconds", "memory": "bytes"}, + "resource_limits": {"cpu_affinity": sorted(cpus) if cpus else None, + "address_space_bytes": args.address_space_bytes, + "scope": "per process; backend fallback service charged separately"}, + "separate_fallback_endpoint": bool(args.fallback_url), + "isolated_baseline_service": bool(args.fallback_url and args.exact_pid and args.fallback_pid and args.exact_pid != args.fallback_pid), "measured_ingest_and_settle_wall_ns": ingest_elapsed, "planning_wall_ns": plan["planning_elapsed_ns"], "process_phases": phases, + "planning_resources": planning_resources, + "storage": storage, + "phase_resources": {name: {service: process_delta(phases[before].get(service), phases[after].get(service)) + for service in PROCESS_IDS} + for name, before, after in [ + ("startup", "startup", "before_ingest"), + ("ingest_and_build", "before_ingest", "after_ingest_and_settle"), + ("queries", "after_ingest_and_settle", "after_queries")]}, "estimated_vs_measured_cost_ratio": None, "acceptance_complete": False, "limitations": ["No common conversion from provider cost units to measured resource units", - "Exact service is shared with fallback; paired order alternates but caches are not isolated", - "Resource budgets must be independently matched; affinity/cgroup recorded, not enforced", + *([] if args.fallback_url else ["Exact service is shared with fallback; caches are not isolated"]), + *([] if cpus else ["CPU affinity is not enforced"]), + "RLIMIT_AS is virtual address space, not physical-memory or aggregate multi-process CPU enforcement", "Raw process RSS is not summary state size; store.json retains backend counters", - "Planning CPU, exact-service construction/storage and isolated cold-start runs remain unmeasured", + "Service startup before supplied PID attachment and isolated cold-cache runs remain unmeasured", "Query equality on one dataset is not a formal approximation confidence guarantee"]} write_json(args.output / "comparison.json", report) finally: diff --git a/tools/o11y-execution/test_compare.py b/tools/o11y-execution/test_compare.py index 800b66a8..b6d3d4e8 100644 --- a/tools/o11y-execution/test_compare.py +++ b/tools/o11y-execution/test_compare.py @@ -43,6 +43,20 @@ def test_failed_or_missing_baseline_prevents_benefit_claim(self): self.assertIsNone(report["matched_query_latency_ratio"]) self.assertIsNone(report["end_to_end_benefit"]) + def test_backend_cpu_includes_separate_fallback_service(self): + """A forwarded query cannot claim CPU savings by omitting Prometheus work.""" + row = {"execution": "exact_fallback", "elapsed_ns": 10, "response": vector(("a", 1)), + "process_resources": {"backend": {"cpu_ns": 2}, "fallback_service": {"cpu_ns": 7}, + "exact_service": {"cpu_ns": 99}}, + "exact": {"http_status": 200, "elapsed_ns": 10, "response": vector(("a", 1)), + "process_resources": {"exact_service": {"cpu_ns": 8}}}} + report = summarize([row]) + self.assertEqual(report["backend_plus_fallback_cpu_ns"], 9) + self.assertEqual(report["baseline_cpu_ns"], 8) + del row["process_resources"]["fallback_service"] + del row["process_resources"]["exact_service"] + self.assertIsNone(summarize([row])["backend_plus_fallback_cpu_ns"]) + def test_matrix_requires_the_same_timestamps_and_empty_series(self): """Series and time coverage are part of correctness even without numeric samples.""" empty = {"status": "success", "data": {"resultType": "matrix", "result": []}} diff --git a/tools/o11y-execution/test_replay.py b/tools/o11y-execution/test_replay.py index 3f51f6da..f1234797 100644 --- a/tools/o11y-execution/test_replay.py +++ b/tools/o11y-execution/test_replay.py @@ -1,7 +1,10 @@ """Behavioral tests for the real-workload replay boundary (not speedup evidence).""" import unittest +from unittest.mock import patch +from tempfile import TemporaryDirectory +from pathlib import Path -from replay import classify, validate_workload, encode_write, parse_samples +from replay import classify, validate_workload, encode_write, parse_samples, replay class ReplayTests(unittest.TestCase): @@ -34,6 +37,16 @@ def test_timestamped_metrics_keep_labels_values_and_time(self): self.assertEqual(rows, [({"__name__": "x_bucket", "le": "1", "job": "a"}, 2.0, 1234)]) self.assertTrue(encode_write(rows)) + def test_odd_corpus_alternates_pair_order_across_repetitions(self): + """An odd corpus must not fix every occurrence to the same pair order.""" + response = {"http_status": 200, "headers": {}, "elapsed_ns": 1, + "response": {"status": "success", "infos": ["data_source: asap_query"], + "data": {"resultType": "vector", "result": []}}} + with TemporaryDirectory() as directory, patch("replay.request", return_value=response): + rows = replay([{"id": "q", "query": "up", "eval_timestamp_ms": 0}], + "http://backend", Path(directory), 2, "http://exact") + self.assertEqual([r["pair_order"] for r in rows], ["exact_first", "backend_first"]) + def test_bad_input_fails_before_any_ingest(self): """No silent sample drops, duplicate samples, or time reordering.""" for lines in [["x NaN 1"], ["x 1"], ["x 1 2", "x 2 1"], ["x 1 1", "x 2 1"]]: