diff --git a/tools/o11y-execution/README.md b/tools/o11y-execution/README.md index deade70f..f235b8e4 100644 --- a/tools/o11y-execution/README.md +++ b/tools/o11y-execution/README.md @@ -57,12 +57,62 @@ classification. Forwarded responses carry a backend-owned `x-asap-execution` header; an unmarked success is not counted as warm or fallback. `ingestion.json` records accepted batches; acceptance does not prove worker completion. -The settle interval is recorded, not a completion barrier. The first traversal is +The runner waits for the finite-input 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 finite-input drain. 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. `--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 +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. A shared fallback/baseline service +can transfer cache warmth; alternating order does not eliminate this confound. ## Finite-input completion diff --git a/tools/o11y-execution/compare.py b/tools/o11y-execution/compare.py new file mode 100644 index 00000000..09d7da22 --- /dev/null +++ b/tools/o11y-execution/compare.py @@ -0,0 +1,137 @@ +"""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) + 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), + "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 a9b58f14..eeb15588 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 @@ -15,6 +17,22 @@ import urllib.parse import urllib.request +from compare import compare_results, process_snapshot, process_delta, summarize + +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": @@ -124,7 +142,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: @@ -135,11 +153,32 @@ def 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} +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 +203,28 @@ 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: + 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 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: + 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 exact_first else "backend_first" write_json(output / "queries.json", rows) return rows @@ -187,6 +237,14 @@ 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("--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=0, help="deprecated; completion uses explicit finite-input drain") parser.add_argument("--repetitions", type=int, default=2) @@ -194,9 +252,32 @@ 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()) + 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) @@ -207,21 +288,37 @@ def main(): "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", "finite-input drain closes trailing panes and permanently seals Remote Write; no live-ingestion claim"]} 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): if child.poll() is not None: @@ -239,16 +336,69 @@ 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") - ingest(samples, [args.exact_url, backend], args.output) + phases["before_ingest"] = process_snapshots() + ingest_start = time.perf_counter_ns() + ingest(samples, list(dict.fromkeys([args.exact_url, fallback_url, backend])), args.output) drained = request(backend + "/api/v1/precompute/drain", b"") write_json(args.output / "drain.json", drained) if drained["http_status"] != 200 or drained["response"].get("complete") is not True: raise RuntimeError("finite-input materialization drain failed; see drain.json") - results = replay(queries, backend, args.output, args.repetitions) - write_json(args.output / "store.json", request(backend + "/api/v1/store/metrics")) + ingest_elapsed = time.perf_counter_ns() - ingest_start + phases["after_ingest_and_drain"] = 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) + 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}) + 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"]}, + "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_drain_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_drain"), + ("queries", "after_ingest_and_drain", "after_queries")]}, + "estimated_vs_measured_cost_ratio": None, + "acceptance_complete": False, + "limitations": ["No common conversion from provider cost units to measured resource units", + *([] 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", + "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: 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..b6d3d4e8 --- /dev/null +++ b/tools/o11y-execution/test_compare.py @@ -0,0 +1,92 @@ +"""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_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": []}} + 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() 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"]]: