diff --git a/README.md b/README.md index 38196adb..d79aee6a 100644 --- a/README.md +++ b/README.md @@ -579,7 +579,10 @@ external-only DAG; only actual summary reads establish ASAP/hybrid execution. The fixture's mixed-DAG assertions are stronger than a successful `SELECT 1`. For recorded datasets, see the [replay guide](docs/user_guide/o11y-replay.md) and -[execution calibration](tools/o11y-execution/CALIBRATION.md). Report correctness, +[execution calibration](tools/o11y-execution/CALIBRATION.md). For synthetic fake +metrics, Google and Alibaba query expressions and cross-engine measurements, see +the [dataset-specific accuracy evaluation](tools/shared-workload/ACCURACY_E2E.md). +Report correctness, fallbacks, build/update cost and whole-deployment resources separately from query latency. Manual examples leave evidence directories intact; remove them only when no longer needed. PID-file cleanup commands apply only to the diff --git a/tools/o11y-execution/calibrate_runtime.py b/tools/o11y-execution/calibrate_runtime.py index 330f0b9b..5587dca6 100644 --- a/tools/o11y-execution/calibrate_runtime.py +++ b/tools/o11y-execution/calibrate_runtime.py @@ -189,9 +189,6 @@ def launch(name, command): if not occurrences: raise RuntimeError(f"no original corpus occurrences for {qid}") exact = {} - for occurrence in occurrences: - params = query_parameters(occurrence, args.disable_result_cache) - exact[occurrence["id"]] = runner._http_request(args.reference_url.rstrip("/") + "/api/v1/query?" + params) records, before, start = [], snapshots(children), time.perf_counter_ns() repeat = 0 measured_cpu = 0 @@ -199,6 +196,8 @@ def launch(name, command): for occurrence in occurrences: params = query_parameters(occurrence, args.disable_result_cache) answer = runner._http_request(query_backend + "/api/v1/query?" + params) + if occurrence["id"] not in exact: + exact[occurrence["id"]] = runner._http_request(args.reference_url.rstrip("/") + "/api/v1/query?" + params) reference = exact[occurrence["id"]] route = runner.classify(answer["response"], answer["headers"]) if answer["http_status"] == 200 else "failed" comparison = compare_results(answer["response"], reference["response"], *comparison_tolerances(occurrence, args)) @@ -232,7 +231,7 @@ def launch(name, command): for child in children.values(): child.terminate() for child in children.values(): - child.wait(timeout=30) + child.wait(timeout=None if args.wait_for_completion else 30) row["resources"]["storage_before_shutdown_bytes"] = row["resources"]["storage_bytes"] row["resources"]["storage_after_shutdown"] = { "exact_bytes": file_bytes(folder / "exact-data"), @@ -255,7 +254,10 @@ def launch(name, command): finally: for child in children.values(): if child.poll() is None: - child.kill() + if args.wait_for_completion: + child.terminate() + else: + child.kill() child.wait() for log in logs: log.close() @@ -390,7 +392,9 @@ def main(): parser.add_argument("--residency-seconds", type=float, default=1) parser.add_argument("--relative-tolerance", type=float, default=0.0) parser.add_argument("--absolute-tolerance", type=float, default=0.0) + parser.add_argument("--wait-for-completion", action="store_true", help="wait without client deadlines or forced shutdown kills") args = parser.parse_args() + runner.HTTP_TIMEOUT = None if args.wait_for_completion else 60 if args.repetitions < 1 or args.max_repetitions < args.repetitions or args.minimum_query_cpu_ns <= 0 or args.residency_seconds < 0: parser.error("positive repetitions and nonnegative residency required") if len({args.backend_port, args.fallback_port, args.metricsql_port}) != 3 or args.exact_cache_bytes <= 0: diff --git a/tools/o11y-execution/process_lifecycle.py b/tools/o11y-execution/process_lifecycle.py index 03a8c0b8..559bcda0 100644 --- a/tools/o11y-execution/process_lifecycle.py +++ b/tools/o11y-execution/process_lifecycle.py @@ -13,7 +13,7 @@ def stop(child, timeout=30): # cannot subsequently be recovered with wait4, and is not a measured zero. if child.returncode is None: child.terminate() - deadline = time.monotonic() + timeout + deadline = time.monotonic() + timeout if timeout is not None else None while True: try: pid, status, collected = os.wait4(child.pid, os.WNOHANG) @@ -23,7 +23,7 @@ def stop(child, timeout=30): child.returncode = os.waitstatus_to_exitcode(status) usage = collected break - if time.monotonic() >= deadline: + if deadline is not None and time.monotonic() >= deadline: forced = True child.kill() try: diff --git a/tools/o11y-execution/replay.py b/tools/o11y-execution/replay.py index a28bd39b..6ba2a1f8 100644 --- a/tools/o11y-execution/replay.py +++ b/tools/o11y-execution/replay.py @@ -95,7 +95,7 @@ def validate_workload(snapshot, corpus): _LABEL = re.compile(r'([a-zA-Z_][a-zA-Z0-9_]*)="((?:[^"\\]|\\[\\"n])*)"') -def iter_samples(lines): +def iter_samples(lines, require_global_order=True): """Strict OpenMetrics subset: seconds converted losslessly to Remote Write milliseconds.""" seen, latest, yielded = {}, -1, False for number, line in enumerate(lines, 1): @@ -123,7 +123,7 @@ def iter_samples(lines): raise ValueError(f"submillisecond timestamp at line {number}") value, timestamp = float(value), int(millis) key = tuple(sorted(labels.items())) - if not math.isfinite(value) or timestamp > 2**63 - 1 or timestamp < latest or timestamp <= seen.get(key, -1): + if not math.isfinite(value) or timestamp > 2**63 - 1 or (require_global_order and timestamp < latest) or timestamp <= seen.get(key, -1): raise ValueError(f"nonfinite, duplicate, or out-of-order sample at line {number}") latest, seen[key] = timestamp, timestamp yielded = True @@ -183,10 +183,13 @@ def encode_write(rows): return varint(length) + literal + wire +HTTP_TIMEOUT = 60 + + 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: + with urllib.request.urlopen(urllib.request.Request(url, data=data, headers=headers or {}), timeout=HTTP_TIMEOUT) as response: body, status, received = response.read(), response.status, dict(response.headers.items()) try: body = json.loads(body) @@ -256,7 +259,7 @@ def ingest(rows, endpoints, output): write_json(output / "ingestion.json", batches) -def replay(queries, backend, output, repetitions, exact_url=None, relative_tolerance=0.0, absolute_tolerance=0.0, evaluation_step_ms=0, batch_resources=False): +def replay(queries, backend, output, repetitions, exact_url=None, relative_tolerance=0.0, absolute_tolerance=0.0, evaluation_step_ms=0, batch_resources=False, backend_first=False): rows = [] if evaluation_step_ms < 0: raise ValueError("evaluation step must be nonnegative") @@ -265,18 +268,30 @@ def replay(queries, backend, output, repetitions, exact_url=None, relative_toler batch_started = time.perf_counter_ns() for repeat in range(repetitions): for query_index, query in enumerate(queries): - exact_first = (repeat + query_index) % 2 == 0 + exact_first = not backend_first and (repeat + query_index) % 2 == 0 evaluation_ms = query["eval_timestamp_ms"] - (repetitions - 1 - repeat) * evaluation_step_ms if evaluation_ms < 0: raise ValueError("advancing evaluation grid predates epoch") params = urllib.parse.urlencode({"query": query["query"], "time": f'{evaluation_ms / 1000:.3f}'}) - # Alternate paired order to expose, rather than always favor, cache/order effects. + # Offline acceptance can require ASAP-first; preserve legacy alternating mode. + def recorded_request(endpoint, engine): + started = time.perf_counter_ns() + try: + answer = query_request(endpoint.rstrip("/") + "/api/v1/query?" + params) + except Exception as error: + answer = {"http_status": None, "response": {"status": "error", "error": str(error)}, + "headers": {}, "elapsed_ns": time.perf_counter_ns() - started} + with (output / "endpoint-requests.jsonl").open("a") as journal: + journal.write(json.dumps({"id": query["id"], "evaluation_ms": evaluation_ms, + "repetition": repeat, "engine": engine, **answer}, allow_nan=False) + "\n") + journal.flush() + return answer exact = None if exact_url and exact_first: - exact = query_request(exact_url.rstrip("/") + "/api/v1/query?" + params) - answer = query_request(backend.rstrip("/") + "/api/v1/query?" + params) + exact = recorded_request(exact_url, "native") + answer = recorded_request(backend, "asap") if exact_url and exact is None: - exact = query_request(exact_url.rstrip("/") + "/api/v1/query?" + params) + exact = recorded_request(exact_url, "native") route = classify(answer["response"], answer["headers"]) if answer["http_status"] != 200: route = "failed" @@ -308,7 +323,29 @@ def replay(queries, backend, output, repetitions, exact_url=None, relative_toler return rows +def verify_summary_ready(queries, backend, output, repetitions, evaluation_step_ms): + """Exercise every admitted window after drain; a warmup is not a timed trial.""" + probes = [] + for repeat in range(repetitions): + for query in queries: + timestamp = query["eval_timestamp_ms"] - (repetitions - 1 - repeat) * evaluation_step_ms + params = urllib.parse.urlencode({"query": query["query"], "time": timestamp / 1000}) + answer = request(backend + "/api/v1/query?" + params) + provenance = execution_provenance(answer["response"], answer["headers"]) + ready = (answer["http_status"] == 200 and classify(answer["response"], answer["headers"]) == "warm" + and (provenance.get("summary_readout_evaluations") or 0) > 0 + and provenance.get("exact_subquery_rpcs") == 0 + and bool(answer["response"].get("data", {}).get("result"))) + probes.append({"id": query["id"], "evaluation_ms": timestamp, "ready": ready, + "provenance": provenance, **answer}) + write_json(output / "summary-readiness.json", {"complete": all(p["ready"] for p in probes), + "scope": "post-drain warm probes at every evaluation window, before measured queries; warms ASAP caches", "probes": probes}) + if not ready: + raise RuntimeError("summary window is not ready; see summary-readiness.json") + + def main(): + global HTTP_TIMEOUT parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--metrics", type=Path, required=True) parser.add_argument("--queries", type=Path, required=True) @@ -332,7 +369,11 @@ def main(): parser.add_argument("--relative-tolerance", type=float, default=0.0) parser.add_argument("--absolute-tolerance", type=float, default=0.0) parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--backend-first", action="store_true") + parser.add_argument("--wait-for-completion", action="store_true", help="no HTTP deadline or forced shutdown kill") + parser.add_argument("--require-summary-ready", action="store_true") args = parser.parse_args() + HTTP_TIMEOUT = None if args.wait_for_completion else 60 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 @@ -438,9 +479,12 @@ def limits(): 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) + if args.require_summary_ready: + verify_summary_ready(queries, backend, args.output, args.repetitions, args.evaluation_step_ms) + phases["after_readiness_probes"] = process_snapshots() results = replay(queries, backend, args.output, args.repetitions, args.exact_url if args.compare else None, args.relative_tolerance, args.absolute_tolerance, - args.evaluation_step_ms, args.batch_resources) + args.evaluation_step_ms, args.batch_resources, args.backend_first) phases["after_queries"] = process_snapshots() write_json(args.output / "process-phases.json", phases) store = request(backend + "/api/v1/store/metrics") @@ -484,7 +528,9 @@ def disk_bytes(path): 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")]}, + *([("readiness_probes", "after_ingest_and_drain", "after_readiness_probes")] + if args.require_summary_ready else []), + ("queries", "after_readiness_probes" if args.require_summary_ready else "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", @@ -502,7 +548,7 @@ def disk_bytes(path): write_json(args.output / "comparison.json", report) finally: from process_lifecycle import stop - write_json(args.output / "backend-lifecycle.json", stop(child, timeout=10)) + write_json(args.output / "backend-lifecycle.json", stop(child, timeout=None if args.wait_for_completion else 10)) if __name__ == "__main__": diff --git a/tools/o11y-execution/run_comparison.py b/tools/o11y-execution/run_comparison.py index bfd4073c..a6cdcbc4 100644 --- a/tools/o11y-execution/run_comparison.py +++ b/tools/o11y-execution/run_comparison.py @@ -29,6 +29,9 @@ def main(): parser.add_argument("--batch-resources", action="store_true") parser.add_argument("--cpu-affinity", required=True) parser.add_argument("--base-port", type=int, default=19410) + parser.add_argument("--backend-first", action="store_true") + parser.add_argument("--wait-for-completion", action="store_true") + parser.add_argument("--require-summary-ready", action="store_true") args = parser.parse_args() if args.trials < 1 or args.repetitions < 1: parser.error("trials and repetitions must be positive") @@ -94,20 +97,30 @@ def main(): command += ["--evaluation-step-ms", str(args.evaluation_step_ms)] if args.batch_resources: command.append("--batch-resources") + if args.backend_first: + command.append("--backend-first") + if args.wait_for_completion: + command.append("--wait-for-completion") + if args.require_summary_ready: + command.append("--require-summary-ready") save(folder / "command.json", command) subprocess.run(command, check=True) finally: cleanup_errors = [] for name, child in children.items(): try: - evidence.setdefault(name, {})["termination"] = stop(child) + evidence.setdefault(name, {})["termination"] = stop(child, timeout=None if args.wait_for_completion else 30) except Exception as error: # An error collecting one service must not orphan the other. evidence.setdefault(name, {})["termination_error"] = repr(error) cleanup_errors.append(f"{name}: {error}") try: - child.kill() - child.wait(timeout=10) + if args.wait_for_completion: + child.terminate() + child.wait() + else: + child.kill() + child.wait(timeout=10) except Exception as kill_error: evidence[name]["cleanup_error"] = repr(kill_error) try: diff --git a/tools/shared-workload/ACCURACY_E2E.md b/tools/shared-workload/ACCURACY_E2E.md new file mode 100644 index 00000000..73fc5f5a --- /dev/null +++ b/tools/shared-workload/ACCURACY_E2E.md @@ -0,0 +1,442 @@ +# Dataset-specific accuracy and cost evaluation + +For developers running controlled experiments. This component binds the shared +PromQL/SQL templates to three datasets, loads identical samples, and compares +Prometheus, ClickHouse, native VictoriaMetrics, ASAP PromQL and ASAP SQL responses +at repeated evaluation timestamps. It records accuracy, HTTP latency and optional +Linux component accounting. It does **not** claim that the backend already +accelerates every expression in the corpus. + +This is an **offline, manually invoked evaluation**. Neither the evaluation nor +its harness tests are registered as a push/PR CI job. Run the commands below +explicitly when preparing an experiment or validating changes to this tooling. +The repository's existing backend CI is unchanged. + +Runtime scope is **backend-local precompute**. No ASAPCollector service is +started or called: the generator sends Remote Write directly to the backend. +The `asap-precompute-rs` build dependency and offline Google mapper source happen +to live in the ASAPCollector repository; they are not an extra running collector. + +## Query matrix + +`accuracy_suite.py manifest` emits **concrete PromQL and ClickHouse SQL** for each +profile, with/without an equality label filter. Only SQL evaluation time +`{eval_ms}` and selector lookback `{lookback_ms}` remain runtime parameters. +Quantiles are 0.5, 0.75, 0.9, 0.95, 0.99; windows are 1m, 10m, 1h, 6h, 24h. +PromQL syntax is `topk by (label_0) (3, metric)` and +`quantile by (label_0) (0.9, metric)`: k/q are arguments, not grouping labels. + +| Family | PromQL shape | Evaluation interval | +|---|---|---| +| 1 | `sum by (G) (M)` | 1s | +| 2 | `topk by (G) (3, M)` | 1s | +| 3 | `quantile by (G) (q, M)` | 1s | +| 4 | `sum_over_time(M[T])` | 1m | +| 5 | `quantile_over_time(q, M[T])` | 1m | +| 6 | `rate(C[T])` | 1m | +| 7 | `sum by (G) (rate(C[T]))` | 1m | +| 8 | `sum by (G) (sum_over_time(M[T]))` | 1m | +| 9 | `topk by (G) (3, rate(C[T]))` | 1m | +| 10 | `quantile_over_time(0.9,M[T]) / quantile_over_time(0.5,M[T])` | 1m | + +Also includes spatial count, count_over_time, increase, nested sum/topk and +temporal-to-spatial compositions. This is a finite sensitivity corpus, **not** +exhaustive coverage of arbitrary-depth SpatialAgg*, all binary operators, all +label matcher operators or vector matching modifiers. `ready` means a template +can be evaluated; it is not a declaration of planner/summary support. + +SQL uses the [shared normalized schema and translations](README.md): latest sample +per complete series identity for spatial queries, `(t-T,t]` temporal bounds, +linearly interpolated quantiles, and reset-corrected, boundary-extrapolated +rate/increase. Rate is not translated as a naive slope. Pin engine versions; +the existing [translation evidence](fixture-results.md) targets Prometheus 3.14.0 +and ClickHouse 26.8.2.7, not all versions. TopK compares labels strictly in this +runner: equally valid tied winners can fail comparison and require review. +Do not interpret these failures as numeric error without inspecting the cutoff. + +## Data selection and concrete expressions + +| Dataset | M (gauge) | G (spatial group) | Series identity | +|---|---|---|---| +| Synthetic | `fake_metric` | `label_0` | label_0, label_1, job, instance | +| Google cluster | `google_cluster_cpu_rate` | `service` | original mapper attributes, including service, task, host | +| Alibaba 2018 | `alibaba_container_cpu_util` | `machine_id` | machine_id, container_id | + +### Synthetic: Prometheus client fake metrics + +This uses Python `prometheus_client` to encode deterministic fake samples with +the metric/label vocabulary from the repository's +[streaming example](../../data_plane/examples/promql/streaming_config.yaml). +It is an offline historical sample generator, **not** a running scraper; 100ms +is the exact sample timestamp interval, not a measured scrape scheduling SLA. +The OpenMetrics file is a historical import stream, not a single `/metrics` +response. Remote Write and SQL import use the matched JSONL rows. + +For group index g, member m and step s, the gauge is +`1 + g % 19 + m + (s % 31)/31`. The separate cumulative +`fake_metric_counter_total` is `(s % 997)*(m+1)` and deliberately resets. +Families 6, 7, 9 and increase use that counter, not the fake gauge. +Examples: `quantile by(label_0)(0.95, fake_metric)` and +`sum by(label_0)(rate(fake_metric_counter_total[10m]))`. + +Vary **group cardinality** over 10, 100, 1,000, 10,000, 100,000, 1,000,000. +Default four member series per group make grouped TopK/quantile nondegenerate. +There are two metrics, hence `2 * groups * members` total series and +`2 * groups * members * (duration_ms/100 + 1)` samples. A 24h, million-group cell +with four members contains **6,912,008,000,000 samples**. Generation is opt-in, +streaming and budget-limited; per-series ordering validation still uses memory +proportional to series cardinality. Do not run the entire matrix on a workstation. +For repeated full-window evaluations, generate history covering T **plus** the +evaluation span, not just T. + +### Google cluster: CPU usage per task, aggregated by job/collection + +Use the existing `ASAPCollector/datasets_eval/google_cluster/otlp_mapper.py` +JSONL export, **not** raw event CSV or scheduling requests. Select records with +`metric == "google_cluster_cpu_rate"`; retain `timestamp_ms`, `value` and all +`attributes`. Require `service`, `task`, `host`; do not drop task identity before +computing spatial quantiles or TopK. + +For 2011 `task_usage`, service is `job-{job_id}`, host is `host-{machine_id}` and +task is the task index. For 2019 `instance_usage`, service is +`coll-{collection_id}` and task is the instance index. This groups usage by the +workload/job, rather than mixing unrelated jobs on a machine. Mapper-created +zone/rack labels are synthetic placement metadata, not measured topology. +Do not enable cardinality-cap hashing: merged identities invalidate accuracy. +See the [official Google trace formats](https://github.com/google/cluster-data). + +Choose a contiguous trace interval and a deterministic set of **complete jobs** +(e.g. sorted job IDs), retaining every task's samples in that interval. Preserve +native timestamps and sampling density; do not upsample to synthetic 100ms or +fill gaps with zeros. Record trace release, source objects/shards, time interval, +selected IDs, mapper revision/options and exclusions alongside the generated +manifest. The generator hashes its input, but cannot infer this provenance from +an already-exported JSONL file. Do not pool 2011 and 2019 runs. + +Examples: `sum by(service)(google_cluster_cpu_rate)` and +`quantile_over_time(0.9,google_cluster_cpu_rate[1h])`. For a filtered manifest use +`--filter-value job-ACTUAL_ID` (2011) or `coll-ACTUAL_ID` (2019); the default +`job-1234567890` is illustrative and must be replaced with a present ID. + +CPU **rate** in the field name means a utilization gauge, not a cumulative +Prometheus counter. Rate/increase families are emitted as `not_applicable` and +skipped. A separately derived integrated counter would be a different experiment +and is not silently manufactured here. + +### Alibaba 2018: container CPU utilization, aggregated by machine + +Use headerless `container_usage.csv`, not `container_meta`, `batch_task`, resource +requests or event counts. The [official 2018 schema](https://github.com/alibaba/clusterdata/blob/master/cluster-trace-v2018/schema.txt) +defines 11 columns. Select column 1 `container_id`, column 2 `machine_id`, +column 3 `time_stamp` (seconds from trace start), column 4 `cpu_util_percent`. +Convert timestamp seconds to integer milliseconds without rounding; preserve +CPU values in source units, with no division by machine core count or invented +counter conversion. The [trace description](https://github.com/alibaba/clusterdata/blob/master/cluster-trace-v2018/trace_2018.md) +explains normalization and collection limitations. + +Choose a contiguous interval and sorted machine IDs, retaining **all containers** +on selected machines. This makes spatial queries compare per-machine container +usage; grouping by container_id alone would often make spatial quantiles trivial. +Record source shard hashes, interval, selected IDs and any cleaning policy. +Negative CPU, the 101 sentinel and nonfinite values fail ingestion; clean them +explicitly and report exclusion counts instead of silently changing the sample +population. Duplicate/out-of-order samples within a series also fail. The tool +does not silently average collisions or sort an arbitrarily large input. + +Examples: `topk by(machine_id)(3, alibaba_container_cpu_util)` and +`sum by(machine_id)(sum_over_time(alibaba_container_cpu_util[6h]))`. +Use `--filter-value m_ACTUAL_ID` for a machine in the selected data. Native +timestamps are retained, including gaps. Configure historical retention at each +receiver; do not independently shift time in different engines. CPU utilization +is a gauge, so counter families are N/A here too. The trace's roughly 4,000 +machines do **not** establish natural million-group coverage; synthetic scale +results must be reported separately from real-trace results. + +## Benefit-scale experiments (offline, opt-in) + +The 10-group, two-minute example below is **only a correctness smoke test**, not +the input size for reporting ASAP benefits. The default `scale_plan.py` profile +is `benefit`; it produces a plan and a window-matched query manifest, not data. + +| Profile | Groups × members per metric | Query T + evaluation span | Samples, both metrics | Unfiltered temporal input samples/query | +|---|---|---|---:|---:| +| smoke | 10 × 4 | 1m + 1m | 96,080 | 24,000 | +| benefit (default) | 1,000 × 16 | 1h + 30m | 1,728,032,000 | 576,000,000 | +| scale | 10,000 × 16 | 6h + 1h | 80,640,320,000 | 34,560,000,000 | +| cardinality | 1,000,000 × 4 | 1m + 1m | 9,608,000,000 | 2,400,000,000 | + +Input counts are logical samples, not measured disk reads: engines may prune, +cache, compress or use indexes. The benefit cell has 32,000 total series, +31 temporal evaluation timestamps at 1m spacing, and 1,801 spatial timestamps at +1s spacing. Full history precedes the **first** query. This tests repeated reuse +of summaries; it does not claim that repetition necessarily amortizes build cost. + +```sh +# Planning is cheap. This writes scale.json, queries.json and a 30-cell matrix.json. +python3 tools/shared-workload/scale_plan.py --profile benefit --output /tmp/benefit-plan +# Only after inspecting scale.json and provisioning storage, explicitly generate: +python3 tools/shared-workload/dataset.py --dataset synthetic \ + --scale-plan /tmp/benefit-plan/scale.json --max-samples 1728032000 \ + --output /srv/eval/benefit-data +``` + +Load the resulting data with `load_dataset.py` as below. Run `accuracy_suite.py run` +with `--manifest /tmp/benefit-plan/queries.json`, its matching `--loaded-data` +receipt, the five endpoints, `--require-warm`, and component accounting. Omit +`--start-ms`/`--end-ms`: the scale-bound manifest supplies the complete planned +repetition interval. The runner rejects undersized/mismatched receipts and +shortened intervals. `--query-name` can select a bounded family experiment. +Queries have **no client deadline** and are never killed for taking too long. +Wait for natural completion. Inspect and record each service's own limits +(query timeout, samples, memory); a service-enforced failure is retained as a +failure, never converted into a speedup or a fabricated completion time. + +The planner also emits the requested six group cardinalities × five windows, +including full repeated-history sizes. It does not automatically execute that +matrix. Use `--groups`, `--members`, `--window` and `--evaluation-minutes` to select +one cell. Sweep members **4, 16, 64 at fixed group count** as a separate axis: +increasing groups alone also increases output size, whereas more members increases +the spatial reduction opportunity at fixed grouped-sum output cardinality. Keep +single-group filtered controls separate: their input does not grow with the number +of unselected groups. Do not describe every query as scanning the full dataset. + +Measure storage in a bounded pilot before generating these datasets. `scale.json` +reports `16 * samples` uncompressed timestamp/value payload (27.65 GB for benefit), +**not** physical disk size: labels, JSONL/OpenMetrics artifacts, indexes, compression, +replicas, WAL and summaries change the actual requirement. Supply +`--measured-bytes-per-sample` from a pilot to estimate **one** store; budget every +baseline/fallback copy plus generated files and temporary storage separately. +No measured bytes/sample means no invented disk estimate. Billion-sample Python +client generation and import can themselves take substantial time; these profiles +are explicit resource commitments, not a promise that the current host can run them. + +Report the crossover curve from smoke through benefit and larger feasible cells, +including cases where ASAP loses. Charge ingestion, materialization, updates and +fallback to ASAP; assess savings over the declared repeated-query horizon, not just +one warm request. Large data is necessary to test scalability, not proof of benefit. +For Google/Alibaba expand contiguous time coverage and complete selected job/machine +populations, report their actual series/sample counts, and preserve native cadence. +Do not replicate trace identities or invent 100ms observations to meet synthetic +targets. A trace subset too small to stress the baseline remains an accuracy control. + +## Generate, load and compare (smoke example) + +```sh +python3 -m pip install -r tools/shared-workload/requirements.txt +python3 tools/shared-workload/dataset.py --dataset synthetic \ + --groups 10 --members 4 --duration-ms 120000 --output /tmp/accuracy-data +python3 tools/shared-workload/accuracy_suite.py manifest \ + --dataset synthetic --output /tmp/accuracy-queries.json +# For traces instead: +# dataset.py --dataset google --input selected-mapper.jsonl --output NEW_DIR +# dataset.py --dataset alibaba --input selected-container_usage.csv --output NEW_DIR +# accuracy_suite.py manifest --dataset google --filter-value job-123 --output NEW_FILE +``` + +Provision **fresh isolated** baseline and fallback stores, with identical versions, +CPU/memory limits, retention, evaluation lookback and data. Create `raw_samples` +using the schema in [README.md](README.md) in an unused ClickHouse database. +Prometheus must enable its Remote Write receiver. Choose historical-retention +settings compatible with the dataset's timestamps. The loader performs writes; +never point it at production or reuse a partially loaded database. + +Configure ASAP before loading: register the manifest's query strings and +`interval_ms` frequencies with the normal control-plane planning workflow, gather +independent calibration costs, compile/install its chosen plan and configure exact +fallbacks. Reuse the [production compiler/replay workflow](../o11y-execution/README.md) +and its [calibration requirements](../o11y-execution/CALIBRATION.md); this component +does not synthesize valid planning snapshots or force a benchmark-selected sketch. +Keep the compiler's candidates, selection, install request/status and versions as +experiment artifacts. An SQL translation may be executable by ClickHouse but not +recognized for summary acceleration by ASAP; such outcomes must remain visible. + +```sh +python3 tools/shared-workload/load_dataset.py --data /tmp/accuracy-data \ + --remote-write http://127.0.0.1:29090/api/v1/write \ + --remote-write http://127.0.0.1:28428/api/v1/write \ + --remote-write http://127.0.0.1:19090/api/v1/write \ + --remote-write http://127.0.0.1:18089/api/v1/write \ + --clickhouse 'http://127.0.0.1:18123/?database=accuracy_fixture' \ + --output /tmp/accuracy-loaded.json +``` + +The endpoints above illustrate Prometheus baseline, VM, a separate Prometheus +fallback, and ASAP ingestion; replace them with actual configured receivers, +including additional SQL-backend ingestion/exact stores if deployed separately. +The receipt means each batch was accepted, **not** that summaries finished. +Invoke the backend's finite-input completion barrier before querying; the existing +replay workflow documents it. Verify all endpoints serve the intended loaded +dataset; receipt/profile/hash checks cannot authenticate arbitrary query URLs. +On partial loading failure, investigate and restart with fresh stores. No retries +or rollback hide partial ingestion. + +```sh +python3 tools/shared-workload/accuracy_suite.py run \ + --manifest /tmp/accuracy-queries.json --loaded-data /tmp/accuracy-loaded.json \ + --query-name spatial_sum --query-name temporal_sum \ + --start-ms 1700000060000 --end-ms 1700000120000 \ + --prometheus http://127.0.0.1:29090 --victoriametrics http://127.0.0.1:28428 \ + --clickhouse 'http://127.0.0.1:18123/?database=accuracy_fixture' \ + --asap-prometheus http://127.0.0.1:18089 \ + --asap-clickhouse http://127.0.0.1:18090 \ + --required-pair promql --required-pair sql \ + --require-warm --output /tmp/accuracy-results.jsonl +python3 tools/shared-workload/summarize_accuracy.py /tmp/accuracy-results.jsonl \ + --output /tmp/accuracy-summary.json +``` + +Repeat `--query-name` to bound a run. With no selection, all ready queries run; +short histories fail full-window checks for larger T. Both filtered and +unfiltered cases are evaluated. Absent filter IDs produce empty-oracle failures. +Results record missing/extra labels and samples, completeness, max absolute and +relative error, zero-baseline/nonfinite differences, tolerances, raw responses, +route evidence and per-engine timings. Defaults are rtol=1e-9, atol=1e-12; choose +and report an approximation budget explicitly, never tune it on evaluation data. +This tolerance is not a statistical sketch guarantee or quantile rank-error bound. + +ClickHouse-vs-Prometheus oracle parity must pass before claiming SQL correctness. +VM's **native** MetricsQL result is compared separately: semantic differences are +reported, not relabeled as sketch error or silently normalized. The manifest also +preserves the separately named VM Prometheus variant for follow-up experiments. +Each comparison pair has its own correctness, execution route and eligibility: +PromQL/Prometheus, SQL/ClickHouse and MetricsQL/VM. Supply `--asap-metricsql` for +the third pair. By default all three pairs are required; a missing MetricsQL +endpoint cannot pass full acceptance. The example explicitly selects only two +pairs using `--required-pair`; its scope does **not** certify VM. Native VM vs +Prometheus is a semantic diagnostic, not a replacement for ASAP MetricsQL vs VM. +VM disagreement with Prometheus can coexist with a valid native-semantics VM pair; +VM disagreement with its paired ASAP endpoint makes that pair ineligible. +Empty oracles, warnings and malformed/duplicate results cannot establish accuracy. +`--require-warm` rejects hybrid, fallback and unknown provenance even when values +match. Without it, a pass means accuracy only, not acceleration. A pair's +`eligible_for_query_comparison` concerns matching warm query service only; +`eligible_for_benefit_conclusion` remains false without full lifecycle evidence. + +This runner is **sequential historical replay**: 1s/1m control evaluation timestamp +spacing, not wall-clock dashboard concurrency. Latency is client-observed HTTP +service time (including failures and connection overhead). **All configured ASAP +endpoints run before native endpoints at each timestamp**. Each request executes +independently; one failure cannot skip the next engine. Completed responses/errors +are immediately flushed to `OUTPUT.endpoints.jsonl`, so an indefinitely slow +later query does not erase earlier evidence. The final output retains all endpoints, +including failures, and summary distributions must not discard unsuccessful cases. +Use independent baseline/fallback stores to prevent ASAP-first fallback traffic +from warming the baseline. Fixed order alone does not isolate caches. This is not +a scheduled live-load or throughput benchmark. + +## Production-planned PromQL chain acceptance + +`planned_run.py` connects one selected corpus query to the existing production +planning/replay workflow. It owns fresh, separate Prometheus baseline/fallback +stores, invokes the normal compiler on measured version-2 cost evidence, installs +its selected artifact, loads data, drains finite-input materialization and probes +**every evaluation window** before timed replay. A readiness probe requires a +nonempty warm response, positive summary-readout count and zero exact-subquery RPCs; +an HTTP-success/warm label alone is insufficient. Probes warm ASAP caches and their +cost is retained separately from query service timing. + +```sh +python3 tools/shared-workload/planned_run.py \ + --data /tmp/accuracy-data --manifest /tmp/accuracy-queries.json \ + --query-id synthetic/1m/False/temporal_sum \ + --snapshot /path/measured-costed-snapshot.json \ + --compiler target/debug/examples/compile_workload_artifact \ + --data-plane target/debug/data_plane --prometheus /path/to/prometheus \ + --cpu-affinity 0,1 --repetitions 2 --output /tmp/planned-sum-acceptance +``` + +Obtain the snapshot via [candidate calibration](../o11y-execution/CALIBRATION.md), +registering exactly the selected query. Use `calibrate_runtime.py +--wait-for-completion` for deadline-free calibration. Candidate calibration does +not manually select the evaluation winner. The driver rejects discovery/demo +snapshots without deployment cost quotes, invalid data hashes and partial windows. +Trace exports may be ordered within each series without being globally ordered. +The driver validates per-series ordering and, when needed, creates a timestamp-sorted +copy using a disk-backed sort. It preserves sample values, labels and timestamps; +`metrics-input.json` records both input hashes. Already ordered input is used directly. +Reserve temporary disk space for the trace copy and sort database. +Its owned replay uses `--backend-first --wait-for-completion +--require-summary-ready`; no client query timeout, subprocess deadline or +timeout-based forced shutdown is used. Owned servers receive normal termination +only after replay returns; the driver waits for shutdown without forced kill. + +`acceptance.json` reports chain correctness, warm/readout evidence, query latency, +planning CPU, observed phase resources, retained storage and artifact locations. +This is **PromQL-only** acceptance, not complete three-engine benefits evaluation. +SQL's moving `{eval_ms}` fixed-plan limitation is not solved by this driver, nor +does it provision a MetricsQL plan. Baseline/fallback stores are distinct, but +their processes overlap in wall time. Independently matched complete lifecycle +runs (including native ingestion and ASAP planning/build/upkeep/retained exact DB) +remain required before a system-benefit claim. No full-cost total is invented. + +The [recorded small sum run](planned-sum-evidence.md) passed this real chain but +ASAP was slower than direct Prometheus in that debug-build fixture. Keep that +negative performance result; do not conflate compiler plan selection with beating +an independently queried native DB. + +## Component and whole-system resources + +Use cgroup v2 with disjoint component cgroups. Supply `--components components.json` +to `accuracy_suite.py run` for request-interval CPU, memory snapshots, block I/O +and optional dedicated-network-namespace counters. Example configuration: + +```json +{ + "asap_promql": { + "data_plane": {"cgroup": "/sys/fs/cgroup/eval/asap", "data_directory": "/srv/eval/asap"}, + "control_plane": {"cgroup": "/sys/fs/cgroup/eval/control"}, + "fallback": {"cgroup": "/sys/fs/cgroup/eval/fallback", "data_directory": "/srv/eval/fallback"} + }, + "asap_sql": {}, + "asap_metricsql": {}, + "prometheus": {"server": {"cgroup": "/sys/fs/cgroup/eval/prom", "data_directory": "/srv/eval/prom"}}, + "clickhouse": {"server": {"cgroup": "/sys/fs/cgroup/eval/ch", "data_directory": "/srv/eval/ch"}}, + "victoriametrics": {"server": {"cgroup": "/sys/fs/cgroup/eval/vm", "data_directory": "/srv/eval/vm"}} +} +``` + +Populate `asap_sql`/`asap_metricsql` with their real components; an empty set +intentionally reports unavailable totals. Include the backend-local data plane, +the planning process during planning, and retained exact fallback services. +Do not include a collector or broker: this profile does not deploy either. +Shared dependencies count once **within** +each system. Reject parent/child cgroup overlap; never sum all five alternative +systems together and call it ASAP cost. `network_namespace_pid` optionally +identifies a distinct non-host namespace per component. Host `/proc/net/dev` is +not per-process accounting. Network RX/TX includes loopback and can count the same +transfer at both endpoints, so it is not summed into a misleading system byte total. + +For a complete phase including ingestion, build, queries and background work: + +```sh +python3 tools/shared-workload/resources.py --components components.json \ + --engine asap_promql --output /tmp/asap-phase.json -- /path/to/experiment-command +``` + +The wrapper runs the explicit command and propagates its exit status. Components +must already exist; run the same phase independently for each baseline. It reports +CPU time, block bytes, per-component and aggregate memory endpoints, sampled memory +peaks (100ms by default), and allocated storage blocks before/after for configured +disjoint data directories. Missing evidence stays null, not zero. Memory is cgroup +memory (including charged cache), not summary heap or process RSS. Sampled peaks +can miss short spikes; directory accounting can be expensive and runs outside the +measured command. The user-supplied component inventory defines the total; the tool +cannot prove no service was omitted. No automatic system-benefit ratio is emitted. + +## Verification and remaining acceptance + +```sh +python3 -m unittest discover -s tools/shared-workload -p 'test_*.py' -v +``` + +Tests cover profile binding, client-encoded samples, trace mappings, duplicate +rejection, all five HTTP adapters, repeated timestamps, fallback/error rejection, +matched loading/hash guards and whole-phase resource/exit-status boundaries. +A local Prometheus 3.5.0 `promtool tsdb create-blocks-from openmetrics` smoke test +imported 32 samples / 16 series spanning 100ms successfully; this verifies import +format only, not the newer query-boundary semantics targeted by the SQL. +HTTP fixtures validate the harness, **not** live engine or planner support; the +separate small PromQL run above establishes one real planned summary-readout chain. +Large trace runs, complete three-engine planned integration and SQL moving-time plans, +all-family warm coverage, production resource measurements, live dashboard load, +quantile rank error and tie-aware TopK accuracy remain separate acceptance work. diff --git a/tools/shared-workload/README.md b/tools/shared-workload/README.md index c2fca3e6..d60a2bad 100644 --- a/tools/shared-workload/README.md +++ b/tools/shared-workload/README.md @@ -1,5 +1,9 @@ # Shared aggregation sensitivity workload +For dataset-specific fake-client, Google and Alibaba expressions, matched HTTP +accuracy/latency replay and component/system resource accounting, see +[Accuracy E2E](ACCURACY_E2E.md). + This developer evaluation tool emits identical timestamped samples for Prometheus, VictoriaMetrics, and ClickHouse, with a query manifest. This synthetic sensitivity study supplements the original o11y workload; it does not replace its coverage or diff --git a/tools/shared-workload/accuracy_suite.py b/tools/shared-workload/accuracy_suite.py new file mode 100644 index 00000000..64974aef --- /dev/null +++ b/tools/shared-workload/accuracy_suite.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Dataset-specific query manifests and repeated HTTP accuracy comparisons.""" +import argparse +import importlib.util +import json +import math +import time +from pathlib import Path +import urllib.parse +import urllib.request +import urllib.error + +from generate import WINDOWS, queries +import resources + +spec = importlib.util.spec_from_file_location( + "accuracy_compare", Path(__file__).resolve().parents[1] / "o11y-execution/compare.py") +comparison = importlib.util.module_from_spec(spec) +spec.loader.exec_module(comparison) + +# Trace profiles use the normalized raw_samples table, not the original CSV. +PROFILES = { + "synthetic": ("fake_metric", "label_0", "g000000", "fake_metric_counter_total"), + "google": ("google_cluster_cpu_rate", "service", "job-1234567890", None), + "alibaba": ("alibaba_container_cpu_util", "machine_id", "m_1", None), +} + + +def corpus(dataset, filter_value=None): + metric, group, default_filter, counter = PROFILES[dataset] + filter_value = default_filter if filter_value is None else filter_value + # Avoid inserting unescaped labels into PromQL or SQL templates. + if not filter_value or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-.:" for c in filter_value): + raise ValueError("filter value must be a nonempty simple trace identifier") + output = [] + for window in WINDOWS: + for filtered in (False, True): + for query in queries(window, filtered): + if query["interval_ms"] == 1000 and window != "1m": + continue # Spatial queries have no T parameter. + is_counter = query["name"].startswith(("rate", "increase")) + selected_metric = counter if is_counter and counter else metric + def bind(expression): + if expression is None: + return None + import re + return re.sub(r"\bdata\b", selected_metric, expression).replace( + "label_0", group).replace("g000000", filter_value) + row = {**query, "id": f"{dataset}/{window}/{filtered}/{query['name']}", + "dataset": dataset, "window": window, + "window_ms": WINDOWS[window], "metric": selected_metric, + "status": "not_applicable" if is_counter and counter is None else "ready", + "reason": "source CPU utilization is a gauge; no native cumulative counter" if is_counter and counter is None else None} + for key in ("promql", "clickhouse_sql", "metricsql", "metricsql_prometheus_variant", "topk_input_promql"): + if key in row: + row[key] = bind(row[key]) + if "topk_group_labels" in row: + row["topk_group_labels"] = [group] + output.append(row) + return {"schema_version": 1, "dataset": dataset, "queries": output, + "cardinalities": [10 ** i for i in range(1, 7)], + "scrape_ms": 100 if dataset == "synthetic" else None, + "trace_sampling": "preserve original timestamps; no implicit resampling", + "coverage": "requested ten families plus representative count, increase and nested compositions; not exhaustive AnyAgg/binary operators"} + + +def request(url, query, timestamp, sql=False, timeout=None): + if sql: + req = urllib.request.Request(url, data=(query + " FORMAT JSONEachRow").encode()) + else: + req = urllib.request.Request(url.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode( + {"query": query, "time": timestamp / 1000})) + with urllib.request.urlopen(req, timeout=timeout) as response: + headers = dict(response.headers.items()) + if sql: + rows = [json.loads(line) for line in response if line.strip()] + body = {"status": "success", "data": {"resultType": "vector", "result": [ + {"metric": row["labels"], "value": [timestamp / 1000, str(row["value"])]} for row in rows]}} + else: + body = json.load(response) + return body, headers + + +def evaluate(actual, expected, rtol, atol): + result = comparison.compare_results(actual, expected, rtol, atol) + _, _, samples = comparison.result_samples(expected) + if not samples: + result.update(equal=False, reason="empty oracle cannot establish accuracy") + return result + + +def execution(body, headers): + h = {k.lower(): v for k, v in headers.items()} + if h.get("x-asap-execution"): + return h["x-asap-execution"] + return "warm" if "data_source: asap_query" in body.get("infos", []) else "unknown" + + +def compare_pair(endpoints, backend, baseline, rtol, atol): + actual, exact = endpoints.get(backend, {}), endpoints.get(baseline, {}) + result = {"backend": backend, "baseline": baseline, + "execution": actual.get("execution", "unavailable"), + "eligible_for_query_comparison": False, + "eligible_for_benefit_conclusion": False, + "full_cost": None, + "cost_reason": "replay has no isolated complete-lifecycle cost evidence"} + if not actual.get("success") or not exact.get("success"): + result["correctness"] = {"equal": False, "comparable": False, + "reason": "paired endpoint missing or failed"} + return result + try: + result["correctness"] = evaluate(actual["response"], exact["response"], rtol, atol) + except (ValueError, KeyError, TypeError) as error: + result["correctness"] = {"equal": False, "comparable": False, "reason": str(error)} + result["eligible_for_query_comparison"] = (result["correctness"]["equal"] and result["execution"] == "warm") + result["latency_ns"] = {"backend": actual["latency_ns"], "baseline": exact["latency_ns"]} + result["query_latency_ratio"] = (exact["latency_ns"] / actual["latency_ns"] + if result["eligible_for_query_comparison"] and actual["latency_ns"] else None) + return result + + +def run(args): + manifest = json.loads(args.manifest.read_text()) + loaded = json.loads(args.loaded_data.read_text())["data"] + scale = manifest.get("scale") + if scale: + if (loaded.get("provenance", {}).get("scale_plan") != scale + or loaded["samples"] != scale["total_samples"] + or loaded["series"] != scale["total_series"]): + raise ValueError("loaded data does not match the requested scale plan") + args.start_ms = scale["evaluation_start_ms"] if args.start_ms is None else args.start_ms + args.end_ms = scale["evaluation_end_ms"] if args.end_ms is None else args.end_ms + if (args.start_ms, args.end_ms) != (scale["evaluation_start_ms"], scale["evaluation_end_ms"]): + raise ValueError("evaluation must cover the planned repetition interval") + if args.start_ms is None or args.end_ms is None or args.end_ms < args.start_ms: + raise ValueError("provide valid start/end timestamps or a scale-bound query manifest") + if loaded.get("provenance", {}).get("dataset") != manifest["dataset"]: + raise ValueError("query and loaded dataset profiles differ") + if args.end_ms > loaded["end_ms"] or args.start_ms < loaded["start_ms"]: + raise ValueError("evaluation outside loaded history") + results = [] + component_sets = json.loads(args.components.read_text()) if args.components else {} + for components in component_sets.values(): + resources.validate(components) + def measured(engine, url, expression, timestamp, sql=False): + components = component_sets.get(engine, {}) + before = resources.snapshot(components) + start = time.perf_counter_ns() + record = {"success": False} + try: + # No client deadline: wait for the service to return, including slow natives. + body, headers = request(url, expression, timestamp, sql, timeout=None) + record.update(success=body.get("status") == "success", response=body, headers=headers, + execution=execution(body, headers) if engine.startswith("asap_") else "native") + if not record["success"]: + record["error"] = {"kind": "api_error", "message": str(body.get("error", body))} + except Exception as error: + record["error"] = {"kind": "timeout" if isinstance(error, TimeoutError) else type(error).__name__, + "message": str(error)} + if isinstance(error, urllib.error.HTTPError): + record["error"].update(http_status=error.code, body=error.read().decode(errors="replace")) + record["headers"] = dict(error.headers.items()) + finally: + record["latency_ns"] = time.perf_counter_ns() - start + record["resources"] = resources.delta(before, resources.snapshot(components)) + return record + with args.output.open("x") as output, args.output.with_name(args.output.name + ".endpoints.jsonl").open("x") as journal: + for query in manifest["queries"]: + if args.query_name and query["name"] not in args.query_name: + continue + if query["status"] != "ready": + continue + for timestamp in range(args.start_ms, args.end_ms + 1, query["interval_ms"]): + row = {"query_id": query["id"], "evaluation_ms": timestamp, + "scale": scale, + "dataset_sha256": loaded["sha256"], "interval_ms": query["interval_ms"], + "schedule": "sequential historical replay, not wall-clock load"} + row["endpoints"] = {} + try: + if query["interval_ms"] != 1000 and timestamp - query["window_ms"] < loaded["start_ms"]: + raise ValueError("full temporal window not present in loaded history") + sql = query["clickhouse_sql"].replace("{eval_ms}", str(timestamp)).replace("{lookback_ms}", "300000") + jobs = [("asap_promql", args.asap_prometheus, query["promql"], False), + ("asap_sql", args.asap_clickhouse, sql, True)] + if getattr(args, "asap_metricsql", None): + jobs.append(("asap_metricsql", args.asap_metricsql, query["metricsql"], False)) + jobs += [("prometheus", args.prometheus, query["promql"], False), + ("clickhouse", args.clickhouse, sql, True), + ("victoriametrics", args.victoriametrics, query["metricsql"], False)] + row["request_order"] = [job[0] for job in jobs] + for engine, url, expression, is_sql in jobs: + record = measured(engine, url, expression, timestamp, is_sql) + row["endpoints"][engine] = record + # Persist each completed request even if the next service never returns. + journal.write(json.dumps({"query_id": query["id"], "evaluation_ms": timestamp, + "engine": engine, **record}, allow_nan=False) + "\n") + journal.flush() + ep = row["endpoints"] + row["pairs"] = {name: compare_pair(ep, backend, baseline, args.rtol, args.atol) + for name, backend, baseline in ( + ("promql", "asap_promql", "prometheus"), + ("sql", "asap_sql", "clickhouse"), + ("metricsql", "asap_metricsql", "victoriametrics"))} + row["oracle_parity"] = compare_pair(ep, "clickhouse", "prometheus", 1e-9, 1e-12)["correctness"] + row["victoriametrics"] = compare_pair(ep, "victoriametrics", "prometheus", 1e-9, 1e-12)["correctness"] + row["sql"], row["promql"] = [row["pairs"][name]["correctness"] for name in ("sql", "promql")] + if not row["oracle_parity"]["equal"]: + row["pairs"]["sql"].update(eligible_for_query_comparison=False, query_latency_ratio=None, + oracle_reason="ClickHouse/Prometheus translation parity unavailable or failed") + row["execution"] = [ep[name].get("execution", "failed") for name in ("asap_promql", "asap_sql")] + required = getattr(args, "required_pairs", None) or ["promql", "sql", "metricsql"] + row["acceptance_scope"] = {"required_pairs": required, "full_benefit_acceptance": False} + row["passed"] = all(row["pairs"][name]["correctness"]["equal"] and + (not args.require_warm or row["pairs"][name]["eligible_for_query_comparison"]) + for name in required) + if "sql" in required: + row["passed"] &= row["oracle_parity"]["equal"] + except Exception as error: + row.update(passed=False, error=str(error)) + row["responses"] = {k: v["response"] for k, v in row["endpoints"].items() if "response" in v} + row["headers"] = {k: v.get("headers", {}) for k, v in row["endpoints"].items()} + row["measurements"] = {k: {"latency_ns": v["latency_ns"], "resources": v["resources"]} + for k, v in row["endpoints"].items()} + results.append(row["passed"]) + output.write(json.dumps(row, allow_nan=False) + "\n") + output.flush() + return 0 if results and all(results) else 1 + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + emit = commands.add_parser("manifest") + emit.add_argument("--dataset", choices=PROFILES, required=True) + emit.add_argument("--filter-value") + emit.add_argument("--output", type=Path, required=True) + check = commands.add_parser("run") + check.add_argument("--manifest", type=Path, required=True) + check.add_argument("--loaded-data", type=Path, required=True) + check.add_argument("--output", type=Path, required=True) + for endpoint in ("prometheus", "victoriametrics", "clickhouse", "asap-prometheus", "asap-clickhouse"): + check.add_argument("--" + endpoint, required=True) + check.add_argument("--asap-metricsql", help="ASAP MetricsQL service paired with native VM; absent means unvalidated VM pair") + check.add_argument("--required-pair", dest="required_pairs", choices=("promql", "sql", "metricsql"), action="append", + help="explicit partial acceptance scope; default requires all three pairs") + check.add_argument("--start-ms", type=int) + check.add_argument("--end-ms", type=int) + check.add_argument("--query-name", action="append", default=[]) + check.add_argument("--rtol", type=float, default=1e-9) + check.add_argument("--atol", type=float, default=1e-12) + check.add_argument("--require-warm", action="store_true") + check.add_argument("--components", type=Path) + args = parser.parse_args() + if args.command == "manifest": + with args.output.open("x") as out: + json.dump(corpus(args.dataset, args.filter_value), out, indent=2) + return 0 + if any(not math.isfinite(v) or v < 0 for v in (args.rtol, args.atol)): + parser.error("tolerances must be finite and nonnegative") + return run(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/shared-workload/dataset.py b/tools/shared-workload/dataset.py new file mode 100644 index 00000000..fcca4a47 --- /dev/null +++ b/tools/shared-workload/dataset.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""Create matched Prometheus-client exposition and ClickHouse rows.""" +import argparse +import csv +from decimal import Decimal +import hashlib +import json +import math +from pathlib import Path + + +def synthetic(groups, members, start_ms, duration_ms): + for step in range(duration_ms // 100 + 1): + for group in range(groups): + for member in range(members): + labels = {"label_0": f"g{group:06d}", "label_1": str(member), + "job": "fake-metrics", "instance": "synthetic:8000"} + ts = start_ms + step * 100 + yield "fake_metric", labels, ts, float(1 + group % 19 + member + (step % 31) / 31) + yield "fake_metric_counter_total", labels, ts, float((step % 997) * (member + 1)) + + +def trace(path, dataset): + with path.open() as source: + if dataset == "google": + # Existing ASAPCollector mapper JSONL preserves original trace identities. + for line in source: + row = json.loads(line) + if row["metric"] != "google_cluster_cpu_rate": + continue + labels = row["attributes"] + if not {"service", "task", "host"} <= labels.keys(): + raise ValueError("Google mapper record is missing service/task/host identity") + ts = Decimal(str(row["timestamp_ms"])) + if ts != int(ts): + raise ValueError("sub-millisecond trace timestamp") + yield row["metric"], labels, int(ts), float(row["value"]) + else: + # Official Alibaba 2018 container_usage.csv has no header. + for row in csv.reader(source): + if len(row) != 11: + raise ValueError("expected 11 columns from Alibaba 2018 container_usage.csv") + value = float(row[3]) + if value < 0 or value == 101 or not math.isfinite(value): + raise ValueError("invalid Alibaba CPU utilization; select/clean source explicitly") + ts = Decimal(row[2]) * 1000 + if ts != int(ts): + raise ValueError("sub-millisecond trace timestamp") + yield "alibaba_container_cpu_util", {"machine_id": row[1], "container_id": row[0]}, int(ts), value + + +def write(root, records, max_samples, provenance=None): + from prometheus_client import CollectorRegistry + from prometheus_client.core import Metric + from prometheus_client.openmetrics.exposition import generate_latest + count, first, last = 0, None, None + latest = {} + root.mkdir(parents=True, exist_ok=False) + with (root / "samples.openmetrics").open("wb") as prom, (root / "samples.jsonl").open("w") as sql: + for metric, labels, ts, value in records: + if count >= max_samples: + raise ValueError("sample budget exceeded; partial output must not be loaded") + if not math.isfinite(value) or not isinstance(ts, int) or ts < 0: + raise ValueError("samples must be finite and have nonnegative integer timestamps") + key = (metric, tuple(sorted(labels.items()))) + if key in latest and ts <= latest[key]: + raise ValueError("duplicate or out-of-order sample within a series") + latest[key] = ts + family = Metric(metric, "Evaluation sample", "unknown") + family.add_sample(metric, labels, value, timestamp=ts / 1000) + class Collector: + def collect(self): + yield family + registry = CollectorRegistry() + registry.register(Collector()) + # Historical sample stream, not repeated HELP/TYPE declarations. + prom.write(b"".join(line + b"\n" for line in generate_latest(registry).splitlines() + if not line.startswith(b"#"))) + sql.write(json.dumps(dict(metric=metric, labels=labels, ts_ms=ts, value=value)) + "\n") + first = min(first, ts) if first is not None else ts + last = max(last, ts) if last is not None else ts + count += 1 + prom.write(b"# EOF\n") + if not count: + raise ValueError("empty dataset") + hashes = {} + for file in (root / "samples.openmetrics", root / "samples.jsonl"): + digest = hashlib.sha256() + with file.open("rb") as data: + for block in iter(lambda: data.read(1024 * 1024), b""): + digest.update(block) + hashes[file.name] = digest.hexdigest() + metadata = dict(samples=count, series=len(latest), start_ms=first, end_ms=last, + sha256=hashes, provenance=provenance or {}) + (root / "data-manifest.json").write_text(json.dumps(metadata, indent=2)) + return metadata + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dataset", choices=("synthetic", "google", "alibaba"), required=True) + parser.add_argument("--input", type=Path) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--groups", type=int, default=10) + parser.add_argument("--members", type=int, default=4) + parser.add_argument("--start-ms", type=int, default=1700000000000) + parser.add_argument("--duration-ms", type=int, default=120000) + parser.add_argument("--max-samples", type=int, default=1000000) + parser.add_argument("--scale-plan", type=Path, help="synthetic scale.json; overrides group/member/time settings, not sample budget") + args = parser.parse_args() + if args.scale_plan: + if args.dataset != "synthetic": + parser.error("synthetic scale plans cannot rescale real traces") + scale = json.loads(args.scale_plan.read_text()) + for name in ("groups", "members", "start_ms", "duration_ms"): + setattr(args, name, scale[name]) + if min(args.groups, args.members, args.max_samples) < 1 or args.duration_ms < 0 or args.duration_ms % 100: + parser.error("counts must be positive and duration nonnegative") + if args.dataset != "synthetic" and not args.input: + parser.error("trace profiles require --input") + if args.dataset == "synthetic" and (args.duration_ms // 100 + 1) * args.groups * args.members * 2 > args.max_samples: + parser.error("synthetic cell exceeds --max-samples; increase budget explicitly") + records = synthetic(args.groups, args.members, args.start_ms, args.duration_ms) if args.dataset == "synthetic" else trace(args.input, args.dataset) + provenance = {"dataset": args.dataset, "generator": "prometheus_client", "groups": args.groups if args.dataset == "synthetic" else None, + "members": args.members if args.dataset == "synthetic" else None, + "scrape_ms": 100 if args.dataset == "synthetic" else None} + if args.scale_plan: + provenance["scale_plan"] = scale + if args.input: + digest = hashlib.sha256() + with args.input.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + provenance.update(input=str(args.input.resolve()), input_sha256=digest.hexdigest()) + print(json.dumps(write(args.output, records, args.max_samples, provenance))) + + +if __name__ == "__main__": + main() diff --git a/tools/shared-workload/load_dataset.py b/tools/shared-workload/load_dataset.py new file mode 100644 index 00000000..e76e2954 --- /dev/null +++ b/tools/shared-workload/load_dataset.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Load the same normalized JSONL samples into isolated evaluation services.""" +import argparse +import hashlib +import json +from pathlib import Path +import sys +import urllib.request + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "o11y-execution")) +from replay import encode_write + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data", type=Path, required=True) + parser.add_argument("--remote-write", action="append", required=True, + help="full write URL; repeat for baseline and ASAP receivers") + parser.add_argument("--clickhouse", required=True, help="HTTP URL including isolated database parameter") + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + if args.output.exists(): + parser.error("receipt already exists; refusing to load again") + # Read and validate the completed manifest before any remote mutations. + manifest = json.loads((args.data / "data-manifest.json").read_text()) + path = args.data / "samples.jsonl" + digest = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + if digest.hexdigest() != manifest["sha256"]["samples.jsonl"]: + raise ValueError("data hash does not match manifest") + def post(url, body, headers=None): + with urllib.request.urlopen(urllib.request.Request(url, data=body, headers=headers or {}), timeout=None) as response: + response.read() + def batch(rows): + encoded = encode_write([({**r["labels"], "__name__": r["metric"]}, r["value"], r["ts_ms"]) for r in rows]) + for endpoint in args.remote_write: + post(endpoint, encoded, {"Content-Type": "application/x-protobuf", "Content-Encoding": "snappy", + "X-Prometheus-Remote-Write-Version": "0.1.0"}) + post(args.clickhouse, ("INSERT INTO raw_samples FORMAT JSONEachRow\n" + "\n".join(json.dumps(r) for r in rows)).encode()) + with path.open() as source: + pending = [] + for line in source: + pending.append(json.loads(line)) + if len(pending) == 1000: + batch(pending) + pending = [] + if pending: + batch(pending) + with args.output.open("x") as out: + json.dump({"data": manifest, "remote_write": args.remote_write, "clickhouse": args.clickhouse, + "scope": "all configured endpoints accepted batches; not proof of summary readiness"}, out, indent=2) + + +if __name__ == "__main__": + main() diff --git a/tools/shared-workload/planned-sum-evidence.md b/tools/shared-workload/planned-sum-evidence.md new file mode 100644 index 00000000..dd19fece --- /dev/null +++ b/tools/shared-workload/planned-sum-evidence.md @@ -0,0 +1,78 @@ +# Small planned PromQL chain: real execution, not benefits acceptance + +On 2026-09-12 the new `planned_run.py` completed one backend-local experiment: +`sum_over_time(fake_metric[1m])`, 10 groups × 4 members, two metrics, 100ms samples, +96,080 total samples / 80 series over two minutes. No ASAPCollector service ran. +Evaluation times were 1700000060000 and 1700000120000 (milliseconds), with complete +1m history for each. Every measured pair executed ASAP first, then Prometheus; +client requests and subprocess execution had no deadlines. + +## Planning and materialization evidence + +The normal candidate exporter produced summary and whole-query fallback candidates. +`calibrate_runtime.py` measured both in isolated runs, `update_global_profile.py` +updated the shared implementation profile, candidates were re-exported/re-measured, +and `calibrate.py` supplied complete measured cost quotes. The production compiler +selected plan **1212003311198828710**, with no candidate override or hand-edited +install artifact. Its projected workload was 1,000 repetitions; the held-out chain +check below used **two** evaluations and is not a validation of that cost horizon. +The exact alternative is backend-forwarded execution, not standalone native DB. + +Plan activation, matched ingestion and finite-input drain succeeded. Both post-drain +readiness probes and both measured ASAP requests reported: + +- execution: warm / detail: asap; +- summary readouts: 1 per request; +- raw scans: 0; +- exact-subquery RPCs: 0. + +Both matched comparisons passed rtol=1e-9 / atol=1e-12, with no missing or extra +series/samples. Maximum relative value error was approximately 3.97e-16; this is +floating-point-tolerant equality, not bitwise identity or a sketch rank guarantee. + +## Observed query timing (only two samples) + +| Endpoint | First | Second | Mean | +|---|---:|---:|---:| +| ASAP | 20.509ms | 26.854ms | 23.682ms | +| Direct Prometheus | 9.067ms | 9.757ms | 9.412ms | + +ASAP was **slower** in this small debug-build run. The native/ASAP query-latency +ratio was about 0.397, not a speedup. Readiness probes warmed ASAP first; native +and fallback used separate empty TSDBs but shared the host and CPU set (0,1). +Two observations, debug binaries and shared-host conditions cannot establish a +performance distribution or extrapolate benefit at scale. Full lifecycle costs +remain unaligned; `eligible_for_full_system_benefit` is false. + +## Reproduction and retained artifacts + +Generate with `dataset.py --dataset synthetic --groups 10 --members 4 +--duration-ms 120000` (default start). Follow the calibration guide and the +`planned_run.py` command in [ACCURACY_E2E.md](ACCURACY_E2E.md). +The local evidence root is `/tmp/pr689-live.pZ4pd1`: + +- `discovery-{candidates.json,measurements/}` and `profile-{candidates.json,measurements/}`: + candidate manifests, raw calibration responses and phase measurements; +- `costed.json`: compiler input with measured quotes; +- `acceptance/run/trial-1/replay/`: planning, install/status, ingestion, drain, + summary readiness, paired queries, phase resources, storage and process lifecycle; +- `acceptance/acceptance.json`: chain decision and limitations. + +Input OpenMetrics SHA256: +`50d5b3d7dd7897a6232e1012692516909769aae066a9e5f1a8a0b161a8913054`. +Costed snapshot SHA256: +`46c1e797e6b2b2f57e0d47f96e4be82dd68970042fe15fabcf4b666dd735d838`. + +Binaries: Prometheus **3.5.0**; backend Rust **1.98.0 debug**, Rust sources from +PR head `2d16f7fb` (subsequent harness-only edits); Planner +`3be523fa0f06a905188e42cbe482d06aa843ba5d`; clean precompute dependency source +`ASAPCollector@9b996305da9f8a2d840cb42714b50d05238a0500`; sketch library at CI's +`8c03d7c68b7150710c79e70a6421971275614561`. The existing dirty dependency trees +were not modified; a temporary dependency path override was reverted after build. +Backend SHA256: `092c24eba9935ae065f84eba28e7dd743b677e503f323ea75e8d80471230d370`. +Compiler SHA256: `0f1dd8817f75b0536697dcbc4a25bf311791a47a6c59b5beae36b814287e8cea`. + +This PromQL run does not validate SQL's Prometheus-3.14 boundary translation, VM +semantics, large-scale benefit, or service-side limit removal. Services retained +their own defaults; no service-side timeout occurred, and all queries completed +naturally. Owned services were then terminated normally, with no forced kill. diff --git a/tools/shared-workload/planned_run.py b/tools/shared-workload/planned_run.py new file mode 100644 index 00000000..2f7c57cd --- /dev/null +++ b/tools/shared-workload/planned_run.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""Run one PromQL acceptance case through the production compiler and owned services.""" +import argparse +from contextlib import closing +from decimal import Decimal +import hashlib +import json +from pathlib import Path +import sqlite3 +import subprocess +import tempfile +import sys + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "o11y-execution")) +from replay import iter_samples + + +def save(path, value): + path.write_text(json.dumps(value, indent=2) + "\n") + + +def sha256(path): + digest = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def prepare(data, manifest, query_id, snapshot, end_ms, repetitions): + metadata = json.loads((data / "data-manifest.json").read_text()) + if metadata.get("provenance", {}).get("dataset") != manifest["dataset"]: + raise ValueError("dataset profile mismatch") + selected = [q for q in manifest["queries"] if q["id"] == query_id and q["status"] == "ready"] + if len(selected) != 1: + raise ValueError("select exactly one ready PromQL query ID") + query = selected[0] + for name, digest in metadata["sha256"].items(): + if sha256(data / name) != digest: + raise ValueError("dataset hash mismatch") + end = metadata["end_ms"] if end_ms is None else end_ms + start = end - (repetitions - 1) * query["interval_ms"] + if repetitions < 1 or end > metadata["end_ms"] or start < metadata["start_ms"]: + raise ValueError("repetition interval exceeds loaded data") + if query["interval_ms"] != 1000 and start - query["window_ms"] < metadata["start_ms"]: + raise ValueError("first query lacks full temporal history") + registered = snapshot["query_workload"] + if registered.get("query_batch") or {q["query"] for q in registered["repeating_queries"]} != {query["promql"]}: + raise ValueError("costed snapshot must register exactly the selected query") + if snapshot.get("snapshot_version") != 2 or not snapshot.get("workload_cost_evidence", {}).get("quotes"): + raise ValueError("measured complete-workload cost evidence is required; discovery/demo costs are not deployment quotes") + return query, {"upstream_revision": "shared-workload:" + hashlib.sha256(json.dumps(manifest, sort_keys=True).encode()).hexdigest(), + "queries": [{"id": query_id, "query": query["promql"], "eval_timestamp_ms": end}]} + + +def prepare_metrics(source, output): + """Keep ordered input unchanged; sort valid interleaved series on disk.""" + ordered, previous = True, -1 + with source.open() as lines: + for _, _, timestamp in iter_samples(lines, require_global_order=False): + ordered = ordered and timestamp >= previous + previous = timestamp + if ordered: + return source + destination = output / "samples-ordered.openmetrics" + # SQLite keeps the sort off the Python heap for historical trace exports. + with tempfile.TemporaryDirectory(prefix="trace-sort-", dir=output) as temporary: + with closing(sqlite3.connect(str(Path(temporary) / "samples.sqlite"))) as database: + database.execute("PRAGMA cache_size=-8192") + database.execute("CREATE TABLE samples (timestamp INTEGER, ordinal INTEGER, line TEXT, PRIMARY KEY (timestamp, ordinal)) WITHOUT ROWID") + def rows(): + with source.open() as lines: + for ordinal, line in enumerate(lines): + if not line.strip() or line.lstrip().startswith("#"): + continue + timestamp = int(Decimal(line.rsplit(None, 1)[1]) * 1000) + yield timestamp, ordinal, line.rstrip("\n") + database.executemany("INSERT INTO samples VALUES (?, ?, ?)", rows()) + database.commit() + with destination.open("x") as target: + for (line,) in database.execute("SELECT line FROM samples ORDER BY timestamp, ordinal"): + target.write(line + "\n") + target.write("# EOF\n") + return destination + + +def acceptance(folder): + replay = folder / "trial-1/replay" + readiness = json.loads((replay / "summary-readiness.json").read_text()) + rows = json.loads((replay / "queries.json").read_text()) + comparison = json.loads((replay / "comparison.json").read_text()) + valid = bool(rows) and readiness["complete"] and all( + r["execution"] == "warm" and r.get("comparison", {}).get("equal") + and r["execution_provenance"].get("summary_readout_evaluations", 0) > 0 + and r["execution_provenance"].get("exact_subquery_rpcs") == 0 + and r.get("pair_order") == "backend_first" for r in rows) + return {"promql_chain_passed": valid, "occurrences": len(rows), "summary_readiness": readiness, + "pair": comparison["all_requests"], "planning_resources": comparison["planning_resources"], + "phase_resources": comparison["phase_resources"], "storage": comparison["storage"], + "isolated_baseline_service": comparison["isolated_baseline_service"], + "eligible_for_full_system_benefit": False, + "limitations": ["PromQL-only chain; SQL moving-time installation and MetricsQL orchestration are not covered", + "owned baseline/fallback stores are distinct, but runs share a host and overlap in wall time", + "resource phases are partial lifecycle evidence, not independently matched full system costs", + "post-drain readiness probes warm ASAP caches before measured requests"], + "artifacts": str(replay.resolve())} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + for name in ("data", "manifest", "snapshot", "compiler", "data-plane", "prometheus", "output"): + parser.add_argument("--" + name, type=Path, required=True) + parser.add_argument("--query-id", required=True) + parser.add_argument("--cpu-affinity", required=True) + parser.add_argument("--repetitions", type=int, default=2) + parser.add_argument("--end-ms", type=int) + parser.add_argument("--base-port", type=int, default=29410) + args = parser.parse_args() + query, corpus = prepare(args.data, json.loads(args.manifest.read_text()), args.query_id, + json.loads(args.snapshot.read_text()), args.end_ms, args.repetitions) + args.output.mkdir(parents=True, exist_ok=False) + metrics = prepare_metrics(args.data / "samples.openmetrics", args.output) + save(args.output / "metrics-input.json", {"source": str((args.data / "samples.openmetrics").resolve()), + "source_sha256": sha256(args.data / "samples.openmetrics"), + "replay_input": str(metrics.resolve()), "replay_sha256": sha256(metrics), + "globally_sorted_copy": metrics != args.data / "samples.openmetrics"}) + corpus_path = args.output / "corpus.json" + save(corpus_path, corpus) + results = args.output / "run" + command = [sys.executable, str(Path(__file__).resolve().parents[1] / "o11y-execution/run_comparison.py"), + "--metrics", str(metrics.resolve()), "--queries", str(corpus_path.resolve()), + "--snapshot", str(args.snapshot.resolve()), "--compiler", str(args.compiler.resolve()), + "--data-plane", str(args.data_plane.resolve()), "--prometheus", str(args.prometheus.resolve()), + "--output", str(results.resolve()), "--cpu-affinity", args.cpu_affinity, + "--base-port", str(args.base_port), "--trials", "1", "--repetitions", str(args.repetitions), + "--evaluation-step-ms", str(query["interval_ms"]), "--backend-first", "--wait-for-completion", "--require-summary-ready"] + save(args.output / "command.json", command) + # No subprocess deadline. Fresh services are owned and shut down only after replay returns. + completed = subprocess.run(command) + if completed.returncode: + save(args.output / "acceptance.json", {"promql_chain_passed": False, "returncode": completed.returncode, + "eligible_for_full_system_benefit": False, "artifacts": str(results.resolve())}) + return completed.returncode + report = acceptance(results) + save(args.output / "acceptance.json", report) + return 0 if report["promql_chain_passed"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/shared-workload/requirements.txt b/tools/shared-workload/requirements.txt new file mode 100644 index 00000000..8816e68a --- /dev/null +++ b/tools/shared-workload/requirements.txt @@ -0,0 +1 @@ +prometheus-client==0.22.1 diff --git a/tools/shared-workload/resources.py b/tools/shared-workload/resources.py new file mode 100644 index 00000000..84d8a4ab --- /dev/null +++ b/tools/shared-workload/resources.py @@ -0,0 +1,148 @@ +"""Linux cgroup-v2 accounting; unavailable evidence remains null.""" +from pathlib import Path +import argparse +import json +import os +import subprocess +import time + + +def validate(components): + roots = [Path(c["cgroup"]).resolve() for c in components.values()] + if len(set(roots)) != len(roots) or any(a in b.parents for a in roots for b in roots if a != b): + raise ValueError("component cgroups must be disjoint, without parent/child overlap") + namespaces = [] + for config in components.values(): + if config.get("network_namespace_pid"): + identity = Path(f"/proc/{int(config['network_namespace_pid'])}/ns/net").stat().st_ino + if identity == Path("/proc/1/ns/net").stat().st_ino or identity in namespaces: + raise ValueError("network accounting requires distinct non-host namespaces") + namespaces.append(identity) + + +def snapshot(components): + result = {} + for name, config in components.items(): + root = Path(config["cgroup"]) + row = {"sampled_ns": time.monotonic_ns()} + try: + row["cpu_usec"] = int(dict(line.split() for line in (root / "cpu.stat").read_text().splitlines())["usage_usec"]) + row["memory_bytes"] = int((root / "memory.current").read_text()) + io = [dict(item.split("=") for item in line.split()[1:]) for line in (root / "io.stat").read_text().splitlines()] + row["disk_read_bytes"] = sum(int(v.get("rbytes", 0)) for v in io) + row["disk_write_bytes"] = sum(int(v.get("wbytes", 0)) for v in io) + row["cgroup_identity"] = root.stat().st_ino + except (OSError, ValueError, KeyError) as error: + row["error"] = str(error) + # A dedicated network namespace is required; host net/dev is not per-process. + row["network"] = None + if config.get("network_namespace_pid"): + try: + net = Path(f"/proc/{int(config['network_namespace_pid'])}/net/dev").read_text().splitlines()[2:] + row["network"] = {"rx_bytes": sum(int(x.split(":")[1].split()[0]) for x in net), + "tx_bytes": sum(int(x.split(":")[1].split()[8]) for x in net)} + except (OSError, ValueError): + pass + result[name] = row + return result + + +def delta(before, after): + rows = {} + for name, initial in before.items(): + end = after[name] + if "error" in initial or "error" in end or initial.get("cgroup_identity") != end.get("cgroup_identity"): + rows[name] = {"error": "component accounting unavailable or cgroup replaced"} + continue + row = {key: end[key] - initial[key] for key in ("cpu_usec", "disk_read_bytes", "disk_write_bytes")} + if any(value < 0 for value in row.values()): + rows[name] = {"error": "accounting counter reset"} + continue + row.update(memory_before_bytes=initial["memory_bytes"], memory_after_bytes=end["memory_bytes"]) + row["network"] = ({k: end["network"][k] - initial["network"][k] for k in ("rx_bytes", "tx_bytes")} + if initial["network"] and end["network"] else None) + rows[name] = row + keys = ("cpu_usec", "disk_read_bytes", "disk_write_bytes", "memory_before_bytes", "memory_after_bytes") + total = {key: sum(row[key] for row in rows.values()) if rows and all(key in row for row in rows.values()) else None for key in keys} + return {"components": rows, "total": total, + "scope": "request interval across listed disjoint cgroups, including background work; endpoint memory snapshots are not peak RSS; network is per dedicated namespace and not summed across links"} + + +def disk_usage(components): + """Allocated filesystem blocks, including retained data; not block-device I/O.""" + result = {} + for name, config in components.items(): + if not config.get("data_directory"): + result[name] = None + continue + try: + root = Path(config["data_directory"]) + if not root.is_dir(): + raise OSError("data directory unavailable") + total, seen = 0, set() + def fail(error): + raise error + for directory, _, files in os.walk(root, onerror=fail): + for path in [Path(directory)] + [Path(directory) / f for f in files]: + st = path.lstat() + identity = (st.st_dev, st.st_ino) + if identity not in seen: + total += st.st_blocks * 512 + seen.add(identity) + result[name] = total + except OSError: + result[name] = None + return result + + +def main(): + parser = argparse.ArgumentParser(description="Measure a complete load/build/query phase around an explicit command") + parser.add_argument("--components", type=Path, required=True) + parser.add_argument("--engine", required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--sample-seconds", type=float, default=0.1) + parser.add_argument("command", nargs=argparse.REMAINDER) + args = parser.parse_args() + command = args.command[1:] if args.command[:1] == ["--"] else args.command + if not command or not 0.01 <= args.sample_seconds <= 60 or args.output.exists(): + parser.error("provide command, unused output, and sample interval in [0.01,60]") + components = json.loads(args.components.read_text())[args.engine] + validate(components) + directories = [Path(c["data_directory"]).resolve() for c in components.values() if c.get("data_directory")] + if len(set(directories)) != len(directories) or any(a in b.parents for a in directories for b in directories if a != b): + parser.error("storage directories must be disjoint") + storage_before = disk_usage(components) + before = snapshot(components) + peaks = {name: row.get("memory_bytes") for name, row in before.items()} + initial_values = list(peaks.values()) + aggregate_peak = sum(initial_values) if initial_values and all(v is not None for v in initial_values) else None + start = time.monotonic_ns() + process = subprocess.Popen(command) + while True: + sample = snapshot(components) + for name, row in sample.items(): + value = row.get("memory_bytes") + peaks[name] = max(peaks[name] or 0, value) if value is not None else peaks[name] + values = [row.get("memory_bytes") for row in sample.values()] + if values and all(v is not None for v in values): + aggregate_peak = max(aggregate_peak or 0, sum(values)) + if process.poll() is not None: + break + time.sleep(args.sample_seconds) + elapsed = time.monotonic_ns() - start + report = delta(before, sample) + report.update(scope="whole wrapped command phase, including ingestion, summary build and idle/background work if included in command", + engine=args.engine, command=command, returncode=process.returncode, elapsed_ns=elapsed, + sampled_memory_peak_bytes=peaks, sampled_total_memory_peak_bytes=aggregate_peak, + sample_seconds=args.sample_seconds, disk_allocated_before_bytes=storage_before, + disk_allocated_after_bytes=disk_usage(components)) + for key in ("disk_allocated_before_bytes", "disk_allocated_after_bytes"): + values = list(report[key].values()) + report["total_" + key] = sum(values) if values and all(v is not None for v in values) else None + with args.output.open("x") as output: + json.dump(report, output, indent=2) + return process.returncode + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/shared-workload/scale_plan.py b/tools/shared-workload/scale_plan.py new file mode 100644 index 00000000..24d98eac --- /dev/null +++ b/tools/shared-workload/scale_plan.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Plan explicit offline scale cells without allocating their datasets.""" +import argparse +import json +import math +from pathlib import Path + +from accuracy_suite import corpus +from generate import WINDOWS + + +# Small correctness fixtures are deliberately distinct from benefit experiments. +PROFILES = { + "smoke": (10, 4, "1m", 1), + "benefit": (1000, 16, "1h", 30), + "scale": (10000, 16, "6h", 60), + "cardinality": (1000000, 4, "1m", 1), +} + + +def plan(profile="benefit", groups=None, members=None, window=None, + evaluation_minutes=None, start_ms=1700000000000, + measured_bytes_per_sample=None): + defaults = PROFILES[profile] + groups = defaults[0] if groups is None else groups + members = defaults[1] if members is None else members + window = defaults[2] if window is None else window + minutes = defaults[3] if evaluation_minutes is None else evaluation_minutes + if min(groups, members, minutes) < 1 or start_ms < 0: + raise ValueError("groups, members and evaluation minutes must be positive; start must be nonnegative") + if measured_bytes_per_sample is not None and (not math.isfinite(measured_bytes_per_sample) or measured_bytes_per_sample <= 0): + raise ValueError("measured bytes per sample must be positive and finite") + window_ms = WINDOWS[window] + span = minutes * 60000 + duration = window_ms + span + series = groups * members + samples = 2 * series * (duration // 100 + 1) + return { + "schema_version": 1, "profile": profile, "dataset": "synthetic", + "purpose": "correctness smoke only" if profile == "smoke" else "candidate benefit experiment; advantage is not guaranteed", + "groups": groups, "members": members, "window": window, "scrape_ms": 100, + "start_ms": start_ms, "duration_ms": duration, + "evaluation_start_ms": start_ms + window_ms, + "evaluation_end_ms": start_ms + duration, + "evaluation_minutes": minutes, "temporal_occurrences_per_query": minutes + 1, + "spatial_occurrences_per_query": span // 1000 + 1, + "series_per_metric": series, "total_series": 2 * series, "total_samples": samples, + "unfiltered_temporal_samples_per_query": series * (window_ms // 100), + "single_group_filtered_temporal_samples_per_query": members * (window_ms // 100), + "spatial_input_series_per_query": series, "spatial_sum_output_groups": groups, + "numeric_payload_bytes_uncompressed": samples * 16, + "measured_store_bytes_per_sample": measured_bytes_per_sample, + "estimated_store_bytes": math.ceil(samples * measured_bytes_per_sample) if measured_bytes_per_sample else None, + "storage_scope": "16-byte timestamp/value payload excludes labels and indexes and is not a disk estimate; optional store estimate uses caller-supplied pilot measurement for one store, not all system copies", + } + + +def query_manifest(scale): + manifest = corpus("synthetic") + manifest["queries"] = [q for q in manifest["queries"] + if q["interval_ms"] == 1000 or q["window"] == scale["window"]] + manifest["scale"] = scale + return manifest + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--profile", choices=PROFILES, default="benefit") + parser.add_argument("--groups", type=int) + parser.add_argument("--members", type=int) + parser.add_argument("--window", choices=WINDOWS) + parser.add_argument("--evaluation-minutes", type=int) + parser.add_argument("--start-ms", type=int, default=1700000000000) + parser.add_argument("--measured-bytes-per-sample", type=float) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + scale = plan(args.profile, args.groups, args.members, args.window, + args.evaluation_minutes, args.start_ms, args.measured_bytes_per_sample) + args.output.mkdir(parents=True, exist_ok=False) + (args.output / "scale.json").write_text(json.dumps(scale, indent=2)) + (args.output / "queries.json").write_text(json.dumps(query_manifest(scale), indent=2)) + # Explicit requested cardinality/window matrix; planning never generates data. + matrix = [plan(args.profile, 10 ** exponent, scale["members"], window, + scale["evaluation_minutes"], args.start_ms, args.measured_bytes_per_sample) + for exponent in range(1, 7) for window in WINDOWS] + (args.output / "matrix.json").write_text(json.dumps(matrix, indent=2)) + print(json.dumps(scale)) + + +if __name__ == "__main__": + main() diff --git a/tools/shared-workload/summarize_accuracy.py b/tools/shared-workload/summarize_accuracy.py new file mode 100644 index 00000000..4472a143 --- /dev/null +++ b/tools/shared-workload/summarize_accuracy.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""Summarize recorded service latency without claiming unmeasured system speedup.""" +import argparse +from collections import defaultdict +import json +from pathlib import Path + +from accuracy_suite import comparison + + +def summarize(rows): + groups = defaultdict(list) + for row in rows: + groups[row["query_id"]].append(row) + result = {} + for query, records in groups.items(): + engines = {engine for row in records for engine in row["measurements"]} + result[query] = { + "occurrences": len(records), "passed": sum(r["passed"] for r in records), + "warm_occurrences": sum(r.get("execution") == ["warm", "warm"] for r in records), + "latency": {engine: comparison.distribution([r["measurements"][engine]["latency_ns"] + for r in records if engine in r["measurements"]]) for engine in sorted(engines)}, + "victoriametrics_equal": sum(r.get("victoriametrics", {}).get("equal", False) for r in records), + "endpoint_failures": {engine: sum(not r.get("endpoints", {}).get(engine, {}).get("success", False) + for r in records) for engine in sorted(engines)}, + "pairs": {}, + } + for name in ("promql", "sql", "metricsql"): + pairs = [r.get("pairs", {}).get(name, {}) for r in records] + eligible = bool(pairs) and all(p.get("eligible_for_query_comparison", False) for p in pairs) + result[query]["pairs"][name] = { + "correct_occurrences": sum(p.get("correctness", {}).get("equal", False) for p in pairs), + "query_comparison_eligible_occurrences": sum(p.get("eligible_for_query_comparison", False) for p in pairs), + "execution_counts": {route: sum(p.get("execution", "unavailable") == route for p in pairs) + for route in ("warm", "hybrid", "exact_fallback", "unknown", "unavailable", "failed")}, + "paired_query_latency_ratio": (sum(p["latency_ns"]["baseline"] for p in pairs) / + sum(p["latency_ns"]["backend"] for p in pairs)) if eligible else None, + "eligible_for_benefit_conclusion": False, + "full_cost": None, + "reason": "complete isolated lifecycle costs are not supplied by endpoint replay", + } + return {"queries": result, "scope": "ASAP-first sequential HTTP latency including failures; no throughput or total-system speedup claim"} + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("input", type=Path) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + with args.input.open() as source: + result = summarize(json.loads(line) for line in source) + with args.output.open("x") as output: + json.dump(result, output, indent=2) diff --git a/tools/shared-workload/test_accuracy.py b/tools/shared-workload/test_accuracy.py new file mode 100644 index 00000000..fddc7a22 --- /dev/null +++ b/tools/shared-workload/test_accuracy.py @@ -0,0 +1,271 @@ +"""Contracts for dataset binding, matched HTTP replay and accounting boundaries.""" +import argparse +from contextlib import contextmanager +import json +from pathlib import Path +import tempfile +import threading +import unittest +from unittest.mock import patch +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + +import accuracy_suite as suite +import dataset +import resources +import load_dataset +from summarize_accuracy import summarize + + +@contextmanager +def endpoints(route="warm", wrong=False): + calls = [] + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + query = parse_qs(urlparse(self.path).query) + calls.append((self.path, query)) + value = "9" if wrong and self.path.startswith("/asap") else "8" + body = {"status": "success", "data": {"resultType": "vector", "result": [ + {"metric": {"label_0": "g000000"}, "value": [float(query["time"][0]), value]}]}} + self.send_response(200) + self.send_header("x-asap-execution", route) + self.end_headers() + self.wfile.write(json.dumps(body).encode()) + + def do_POST(self): + body = self.rfile.read(int(self.headers["Content-Length"])) + calls.append((self.path, body)) + self.send_response(200) + self.send_header("x-asap-execution", route) + self.end_headers() + self.wfile.write(b'{"labels":{"label_0":"g000000"},"value":8}\n') + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}", calls + finally: + server.shutdown() + server.server_close() + thread.join() + + +class AccuracyTests(unittest.TestCase): + def test_profiles_and_families(self): + """Every requested window/filter is bound, while gauge counters are N/A.""" + for profile, (metric, group, _, counter) in suite.PROFILES.items(): + manifest = suite.corpus(profile) + queries = manifest["queries"] + self.assertEqual(len({q["id"] for q in queries}), len(queries)) + self.assertEqual({q["window"] for q in queries}, set(suite.WINDOWS)) + self.assertEqual(manifest["cardinalities"], [10, 100, 1000, 10000, 100000, 1000000]) + for q in queries: + expected = counter or metric if q["name"].startswith(("rate", "increase")) else metric + self.assertIn(expected, q["promql"]) + self.assertIn(expected, q["clickhouse_sql"]) + self.assertNotIn("'data'", q["clickhouse_sql"]) + if profile != "synthetic": + self.assertNotIn("label_0", q["promql"]) + if q["name"].startswith(("rate", "increase")): + self.assertEqual(q["status"], "not_applicable") + for name in ("spatial_sum", "spatial_topk", "spatial_quantile_0.99", "temporal_sum", + "temporal_quantile_0.75", "rate", "rate_spatial_sum", "temporal_sum_spatial_sum", + "rate_spatial_topk", "quantile_ratio", "spatial_count", "increase", "nested_spatial_sum"): + self.assertTrue(any(q["name"] == name for q in queries), name) + with self.assertRaises(ValueError): + suite.corpus("google", "x' OR 1=1") + + def test_client_data_and_counter_resets(self): + """The client exposition and SQL rows encode identical 100ms samples.""" + from prometheus_client.openmetrics.parser import text_string_to_metric_families + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) / "data" + meta = dataset.write(root, dataset.synthetic(2, 4, 1000000, 100), 32) + rows = [json.loads(line) for line in (root / "samples.jsonl").read_text().splitlines()] + # A historical import stream repeats families; unlike one scrape, + # parse its individual client-encoded sample lines independently. + samples = [s for line in (root / "samples.openmetrics").read_text().splitlines() if not line.startswith("#") + for family in text_string_to_metric_families(line + "\n# EOF\n") for s in family.samples] + self.assertEqual(meta["samples"], 32) + self.assertEqual(meta["series"], 16) + self.assertEqual(len(samples), len(rows)) + for sample, row in zip(samples, rows): + self.assertEqual((sample.name, sample.labels, sample.value, float(sample.timestamp) * 1000), + (row["metric"], row["labels"], row["value"], row["ts_ms"])) + values = [v for m, _, _, v in dataset.synthetic(1, 1, 0, 99800) if m.endswith("_total")] + self.assertGreater(values[996], values[997]) + + def test_trace_mapping_and_duplicates(self): + """Native timestamps and identities survive normalization; duplicates fail.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + google = root / "google.jsonl" + google.write_text(json.dumps({"metric": "google_cluster_cpu_rate", "timestamp_ms": 1234, + "value": 0.25, "attributes": {"service": "job-1", "host": "host-2", "task": "3"}}) + "\n") + self.assertEqual(list(dataset.trace(google, "google"))[0][2:], (1234, 0.25)) + alibaba = root / "container_usage.csv" + alibaba.write_text("c_1,m_1,12.5,25,30,0,0,0,0,0,0\n") + row = list(dataset.trace(alibaba, "alibaba"))[0] + self.assertEqual(row, ("alibaba_container_cpu_util", {"machine_id": "m_1", "container_id": "c_1"}, 12500, 25.0)) + with self.assertRaisesRegex(ValueError, "duplicate"): + dataset.write(root / "bad", [row, row], 10) + self.assertFalse((root / "bad/data-manifest.json").exists()) + + def run_fixture(self, root, url, **changes): + manifest = suite.corpus("synthetic") + manifest["queries"] = [q for q in manifest["queries"] if q["name"] == "spatial_sum" and not q["filtered"]] + (root / "queries.json").write_text(json.dumps(manifest)) + (root / "loaded.json").write_text(json.dumps({"data": {"start_ms": 1000000, "end_ms": 1002000, + "provenance": {"dataset": "synthetic"}, "sha256": {"samples.jsonl": "fixture"}}})) + args = argparse.Namespace(manifest=root / "queries.json", loaded_data=root / "loaded.json", output=root / "results.jsonl", + prometheus=url + "/prom", victoriametrics=url + "/vm", clickhouse=url + "/ch", + asap_prometheus=url + "/asap", asap_clickhouse=url + "/asap-sql", + start_ms=1000000, end_ms=1002000, query_name=[], rtol=1e-9, atol=1e-12, + require_warm=True, components=None, required_pairs=["promql", "sql"]) + for key, value in changes.items(): + setattr(args, key, value) + code = suite.run(args) + return code, [json.loads(line) for line in args.output.read_text().splitlines()] + + def test_http_e2e_schedule_and_sql_rendering(self): + """All five HTTP adapters execute matched timestamps and record measurements.""" + with tempfile.TemporaryDirectory() as tmp, endpoints() as (url, calls): + code, rows = self.run_fixture(Path(tmp), url) + self.assertEqual(code, 0) + self.assertEqual(len(calls), 15) + self.assertEqual([r["evaluation_ms"] for r in rows], [1000000, 1001000, 1002000]) + self.assertEqual(len(rows[0]["measurements"]), 5) + sql = next(body for path, body in calls if path == "/ch") + self.assertNotIn(b"{eval_ms}", sql) + self.assertTrue(sql.endswith(b"FORMAT JSONEachRow")) + summary = next(iter(summarize(rows)["queries"].values())) + self.assertEqual(summary["warm_occurrences"], 3) + self.assertEqual(summary["latency"]["prometheus"]["count"], 3) + + def test_wrong_results_and_fallback_fail(self): + """Equality cannot disguise fallback and warm evidence cannot disguise error.""" + for route, wrong in (("exact_fallback", False), ("unknown", False), ("warm", True)): + with tempfile.TemporaryDirectory() as tmp, endpoints(route, wrong) as (url, _): + code, rows = self.run_fixture(Path(tmp), url) + self.assertEqual(code, 1) + self.assertFalse(any(row["passed"] for row in rows)) + + def test_native_timeout_does_not_skip_other_endpoints(self): + """A failed VM call cannot erase successful responses or skip ASAP calls.""" + original = suite.request + for failed in ("/prom", "/ch", "/vm"): + with tempfile.TemporaryDirectory() as tmp, endpoints() as (url, calls): + def probe(endpoint, *args, **kwargs): + if endpoint == url + failed: + raise TimeoutError("native timed out") + return original(endpoint, *args, **kwargs) + with patch.object(suite, "request", side_effect=probe): + _, rows = self.run_fixture(Path(tmp), url) + self.assertEqual(sum(path.startswith("/asap") for path, _ in calls), 6) + self.assertTrue(all("asap_promql" in row["responses"] for row in rows)) + engine = {"/prom": "prometheus", "/ch": "clickhouse", "/vm": "victoriametrics"}[failed] + self.assertTrue(all(row["endpoints"][engine]["error"]["kind"] == "timeout" for row in rows)) + + def test_vm_mismatch_cannot_establish_vm_benefit(self): + """Native VM semantics and a missing ASAP MetricsQL pair are explicit.""" + original = suite.request + with tempfile.TemporaryDirectory() as tmp, endpoints() as (url, _): + def probe(endpoint, *args, **kwargs): + body, headers = original(endpoint, *args, **kwargs) + if endpoint == url + "/vm": + body["data"]["result"][0]["value"][1] = "999" + return body, headers + with patch.object(suite, "request", side_effect=probe): + code, rows = self.run_fixture(Path(tmp), url, asap_metricsql=url + "/asap-vm", + required_pairs=["promql", "sql", "metricsql"]) + self.assertEqual(code, 1) + self.assertTrue(all(not row["pairs"]["metricsql"]["eligible_for_query_comparison"] for row in rows)) + + def test_missing_metricsql_pair_is_not_full_acceptance(self): + """Without an ASAP VM endpoint the default three-pair acceptance fails.""" + with tempfile.TemporaryDirectory() as tmp, endpoints() as (url, _): + code, rows = self.run_fixture(Path(tmp), url, required_pairs=None) + self.assertEqual(code, 1) + self.assertTrue(all(not row["pairs"]["metricsql"]["correctness"]["comparable"] for row in rows)) + + def test_asap_first_without_client_deadline_and_vm_pair(self): + """ASAP runs first without a deadline; MetricsQL has its own exact pair.""" + original = suite.request + with tempfile.TemporaryDirectory() as tmp, endpoints() as (url, calls): + def probe(endpoint, *args, **kwargs): + self.assertIsNone(kwargs["timeout"]) + return original(endpoint, *args, **kwargs) + with patch.object(suite, "request", side_effect=probe): + code, rows = self.run_fixture(Path(tmp), url, asap_metricsql=url + "/asap-vm", + required_pairs=["promql", "sql", "metricsql"]) + self.assertEqual(code, 0) + self.assertTrue(all(row["pairs"]["metricsql"]["eligible_for_query_comparison"] for row in rows)) + self.assertTrue(all(not row["pairs"]["metricsql"]["eligible_for_benefit_conclusion"] for row in rows)) + self.assertEqual(rows[0]["request_order"], ["asap_promql", "asap_sql", "asap_metricsql", "prometheus", "clickhouse", "victoriametrics"]) + journal = Path(tmp) / "results.jsonl.endpoints.jsonl" + self.assertEqual(len(journal.read_text().splitlines()), 18) + + def test_empty_oracle_fails(self): + """An empty result pair is not positive accuracy evidence.""" + empty = {"status": "success", "data": {"resultType": "vector", "result": []}} + self.assertFalse(suite.evaluate(empty, empty, 0, 0)["equal"]) + + def test_resources_no_double_counting(self): + """Overlapping cgroups are rejected and absent accounting stays null.""" + with self.assertRaises(ValueError): + resources.validate({"a": {"cgroup": "/sys/fs/cgroup/test"}, "b": {"cgroup": "/sys/fs/cgroup/test/child"}}) + self.assertIsNone(resources.delta({}, {})["total"]["cpu_usec"]) + sample = {"x": {"cgroup_identity": 1, "cpu_usec": 10, "disk_read_bytes": 2, + "disk_write_bytes": 4, "memory_bytes": 100, "network": None}} + after = {"x": {**sample["x"], "cpu_usec": 20, "memory_bytes": 150}} + result = resources.delta(sample, after) + self.assertEqual(result["total"]["cpu_usec"], 10) + self.assertEqual(result["total"]["memory_after_bytes"], 150) + + def test_loader_http_receipt_and_hash_guard(self): + """One source feeds Remote Write and SQL; tampering fails before writes.""" + with tempfile.TemporaryDirectory() as tmp, endpoints() as (url, calls): + root = Path(tmp) + dataset.write(root / "data", dataset.synthetic(1, 1, 1000000, 100), 4, + {"dataset": "synthetic"}) + argv = ["load_dataset.py", "--data", str(root / "data"), "--remote-write", url + "/write", + "--clickhouse", url + "/ch", "--output", str(root / "receipt.json")] + with patch("sys.argv", argv): + load_dataset.main() + self.assertEqual(len(calls), 2) + self.assertIsInstance(calls[0][1], bytes) + self.assertTrue(calls[1][1].startswith(b"INSERT INTO raw_samples FORMAT JSONEachRow\n")) + self.assertEqual(json.loads((root / "receipt.json").read_text())["data"]["samples"], 4) + (root / "data/samples.jsonl").write_text("tampered\n") + argv[-1] = str(root / "second-receipt.json") + with patch("sys.argv", argv), self.assertRaisesRegex(ValueError, "hash"): + load_dataset.main() + self.assertEqual(len(calls), 2) + + def test_phase_wrapper_preserves_exit_and_totals(self): + """Whole-command measurement reports disk/CPU and preserves failure status.""" + import sys + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + cgroup = root / "cgroup" + cgroup.mkdir() + (cgroup / "cpu.stat").write_text("usage_usec 100\n") + (cgroup / "memory.current").write_text("500\n") + (cgroup / "io.stat").write_text("8:0 rbytes=10 wbytes=20\n") + config = root / "components.json" + config.write_text(json.dumps({"asap_promql": {"backend": {"cgroup": str(cgroup), "data_directory": str(cgroup)}}})) + output = root / "phase.json" + with patch("sys.argv", ["resources.py", "--components", str(config), "--engine", "asap_promql", + "--output", str(output), "--", sys.executable, "-c", "raise SystemExit(3)"]): + self.assertEqual(resources.main(), 3) + report = json.loads(output.read_text()) + self.assertEqual(report["total"]["cpu_usec"], 0) + self.assertEqual(report["sampled_total_memory_peak_bytes"], 500) + self.assertGreater(report["total_disk_allocated_after_bytes"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/shared-workload/test_planned.py b/tools/shared-workload/test_planned.py new file mode 100644 index 00000000..6b5bb327 --- /dev/null +++ b/tools/shared-workload/test_planned.py @@ -0,0 +1,126 @@ +"""Preflight and readiness contracts for production-planned PromQL acceptance.""" +import json +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +import accuracy_suite +import dataset +import load_dataset # Makes the existing replay module available. +import replay +import planned_run +from planned_run import prepare + + +class PlannedTests(unittest.TestCase): + def test_series_ordered_trace_reaches_production_replay(self): + """Globally interleaved trace times remain unchanged and replay in time order.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + records = [("google_cluster_cpu_rate", {"service": "job-a", "task": task, "host": "h"}, ts, value) + for task, samples in [("a", [(0, 1.0), (120000, 2.0)]), + ("b", [(0, 3.0), (120000, 4.0)])] + for ts, value in samples] + dataset.write(root / "data", records, 4, {"dataset": "google"}) + original = (root / "data/samples.openmetrics").read_bytes() + manifest = accuracy_suite.corpus("google") + (root / "queries.json").write_text(json.dumps(manifest)) + (root / "snapshot.json").write_text(json.dumps({ + "snapshot_version": 2, + "query_workload": {"repeating_queries": [{"query": "sum_over_time(google_cluster_cpu_rate[1m])"}]}, + "workload_cost_evidence": {"quotes": [{"test_placeholder": True}]}})) + argv = ["planned_run.py", "--data", str(root / "data"), "--manifest", str(root / "queries.json"), + "--snapshot", str(root / "snapshot.json"), "--query-id", "google/1m/False/temporal_sum", + "--compiler", "compiler", "--data-plane", "backend", "--prometheus", "prometheus", + "--cpu-affinity", "0", "--output", str(root / "run")] + def replay_command(command): + path = Path(command[command.index("--metrics") + 1]) + with path.open() as source: + samples = list(replay.iter_samples(source)) + self.assertEqual([row[2] for row in samples], [0, 0, 120000, 120000]) + actual = {(labels["task"], timestamp, value) for labels, value, timestamp in samples} + self.assertEqual(actual, {(labels["task"], timestamp, value) + for _, labels, timestamp, value in records}) + return type("Completed", (), {"returncode": 1})() + with patch("sys.argv", argv), patch.object(planned_run.subprocess, "run", side_effect=replay_command): + self.assertEqual(planned_run.main(), 1) + self.assertEqual((root / "data/samples.openmetrics").read_bytes(), original) + + def test_ordered_metrics_use_original_file(self): + """Ordered synthetic data needs no copy or timestamp changes.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "samples.openmetrics" + source.write_text('x{task="a"} 1 0\nx{task="b"} 2 0\nx{task="a"} 3 0.1\n# EOF\n') + self.assertEqual(planned_run.prepare_metrics(source, root), source) + self.assertEqual(list(root.iterdir()), [source]) + + def test_sort_does_not_hide_invalid_per_series_order(self): + """Sorting must not repair duplicate or backwards samples within a series.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "samples.openmetrics" + for timestamp in ["0", "0.1"]: + source.write_text(f'x{{task="a"}} 1 0.1\nx{{task="b"}} 2 0\nx{{task="a"}} 3 {timestamp}\n') + with self.assertRaisesRegex(ValueError, "out-of-order"): + planned_run.prepare_metrics(source, root) + self.assertEqual(list(root.iterdir()), [source]) + + def test_complete_window_and_costed_registration_are_required(self): + """Reject unpriced plans and repetitions whose first window predates the data.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + dataset.write(root / "data", dataset.synthetic(1, 1, 0, 120000), 2402, {"dataset": "synthetic"}) + manifest = accuracy_suite.corpus("synthetic") + snapshot = {"snapshot_version": 2, "query_workload": {"repeating_queries": [{"query": "sum_over_time(fake_metric[1m])"}]}} + qid = "synthetic/1m/False/temporal_sum" + with self.assertRaisesRegex(ValueError, "cost evidence"): + prepare(root / "data", manifest, qid, snapshot, None, 2) + snapshot["workload_cost_evidence"] = {"quotes": [{"test_placeholder": True}]} + _, corpus = prepare(root / "data", manifest, qid, snapshot, None, 2) + self.assertEqual(corpus["queries"][0]["eval_timestamp_ms"], 120000) + with self.assertRaisesRegex(ValueError, "full temporal history"): + prepare(root / "data", manifest, qid, snapshot, None, 3) + + def test_every_window_needs_observed_summary_readout(self): + """A warm label without a real summary read is insufficient readiness evidence.""" + body = {"status": "success", "infos": ["data_source: asap_query"], + "data": {"resultType": "vector", "result": [{"metric": {}, "value": [120, "8"]}]}} + headers = {"x-asap-summary-readout-evaluations": "1", "x-asap-exact-subquery-rpcs": "0"} + answer = {"http_status": 200, "response": body, "headers": headers} + query = [{"id": "q", "query": "sum_over_time(fake_metric[1m])", "eval_timestamp_ms": 120000}] + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + with patch.object(replay, "request", return_value=answer) as calls: + replay.verify_summary_ready(query, "http://asap", root, 2, 60000) + self.assertEqual(calls.call_count, 2) + report = json.loads((root / "summary-readiness.json").read_text()) + self.assertTrue(report["complete"]) + self.assertEqual([p["evaluation_ms"] for p in report["probes"]], [60000, 120000]) + with patch.object(replay, "request", return_value={**answer, "headers": {}}), self.assertRaisesRegex(RuntimeError, "not ready"): + replay.verify_summary_ready(query, "http://asap", root, 2, 60000) + + def test_planned_replay_journals_asap_before_waiting_for_native(self): + """The production replay also preserves ASAP evidence before a native failure.""" + answer = {"http_status": 200, "response": {"status": "success", "infos": ["data_source: asap_query"], + "data": {"resultType": "vector", "result": [{"metric": {}, "value": [120, "8"]}]}}, + "headers": {}, "elapsed_ns": 100} + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + def request(url): + if url.startswith("http://native"): + completed = [json.loads(line) for line in (root / "endpoint-requests.jsonl").read_text().splitlines()] + self.assertEqual(completed[0]["engine"], "asap") + raise OSError("service disconnected") + return answer + queries = [{"id": "q", "query": "sum_over_time(fake_metric[1m])", "eval_timestamp_ms": 120000}] + with patch.object(replay, "request", side_effect=request): + rows = replay.replay(queries, "http://asap", root, 1, exact_url="http://native", backend_first=True) + self.assertEqual(rows[0]["execution"], "warm") + self.assertEqual(rows[0]["exact"]["response"]["status"], "error") + self.assertFalse(rows[0]["comparison"]["equal"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/shared-workload/test_scale.py b/tools/shared-workload/test_scale.py new file mode 100644 index 00000000..bf7f49a7 --- /dev/null +++ b/tools/shared-workload/test_scale.py @@ -0,0 +1,116 @@ +"""Scale planning never allocates the planned billion-sample datasets in tests.""" +import argparse +import json +import io +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +import accuracy_suite +import dataset +from scale_plan import main as plan_main, plan, query_manifest +from test_accuracy import endpoints + + +class ScaleTests(unittest.TestCase): + def test_benefit_is_large_and_has_complete_repeated_windows(self): + """The default benefit cell has billions of samples, not smoke-size input.""" + scale = plan() + self.assertEqual(scale["total_samples"], 1728032000) + self.assertEqual(scale["total_series"], 32000) + self.assertEqual(scale["unfiltered_temporal_samples_per_query"], 576000000) + self.assertEqual(scale["temporal_occurrences_per_query"], 31) + self.assertEqual(scale["spatial_occurrences_per_query"], 1801) + self.assertEqual(scale["evaluation_start_ms"] - scale["start_ms"], 3600000) + self.assertEqual(scale["evaluation_end_ms"] - scale["evaluation_start_ms"], 1800000) + self.assertIsNone(scale["estimated_store_bytes"]) + self.assertTrue(all(q["interval_ms"] == 1000 or q["window"] == "1h" + for q in query_manifest(scale)["queries"])) + + def test_members_and_cardinality_are_distinct_scale_axes(self): + """Increasing members increases scan work without changing sum output groups.""" + small, large = plan(members=4), plan(members=64) + self.assertEqual(small["spatial_sum_output_groups"], large["spatial_sum_output_groups"]) + self.assertEqual(large["unfiltered_temporal_samples_per_query"], 16 * small["unfiltered_temporal_samples_per_query"]) + self.assertEqual(plan("cardinality")["total_series"], 8000000) + self.assertEqual(plan(measured_bytes_per_sample=20)["estimated_store_bytes"], 1728032000 * 20) + + def test_scale_generation_requires_explicit_budget(self): + """A large plan fails before creating output unless its sample budget is allowed.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + (root / "scale.json").write_text(json.dumps(plan())) + with patch("sys.argv", ["dataset.py", "--dataset", "synthetic", "--scale-plan", str(root / "scale.json"), + "--output", str(root / "data")]), patch("sys.stderr", io.StringIO()), self.assertRaises(SystemExit): + dataset.main() + self.assertFalse((root / "data").exists()) + + def test_plan_command_emits_matrix_without_allocating_samples(self): + """All requested cardinality/window cells are planned without generating data.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) / "plan" + with patch("sys.argv", ["scale_plan.py", "--output", str(root)]), patch("builtins.print"): + plan_main() + self.assertEqual({p.name for p in root.iterdir()}, {"scale.json", "queries.json", "matrix.json"}) + matrix = json.loads((root / "matrix.json").read_text()) + self.assertEqual(len(matrix), 30) + self.assertEqual({c["groups"] for c in matrix}, {10, 100, 1000, 10000, 100000, 1000000}) + + def test_runner_uses_planned_timestamps(self): + """A matching scale receipt supplies the full interval to the five HTTP adapters.""" + with tempfile.TemporaryDirectory() as tmp, endpoints() as (url, calls): + root = Path(tmp) + scale = plan("smoke", groups=1, members=1) + (root / "queries.json").write_text(json.dumps(query_manifest(scale))) + meta = {"provenance": {"dataset": "synthetic", "scale_plan": scale}, + "samples": scale["total_samples"], "series": scale["total_series"], + "start_ms": scale["start_ms"], "end_ms": scale["evaluation_end_ms"], "sha256": {}} + (root / "receipt.json").write_text(json.dumps({"data": meta})) + args = argparse.Namespace(manifest=root / "queries.json", loaded_data=root / "receipt.json", output=root / "results.jsonl", + start_ms=None, end_ms=None, query_name=["temporal_sum"], components=None, + prometheus=url + "/prom", victoriametrics=url + "/vm", clickhouse=url + "/ch", + asap_prometheus=url + "/asap", asap_clickhouse=url + "/asap-sql", + rtol=1e-9, atol=1e-12, require_warm=True, required_pairs=["promql", "sql"]) + self.assertEqual(accuracy_suite.run(args), 0) + rows = [json.loads(line) for line in args.output.read_text().splitlines()] + self.assertEqual(len(calls), 20) + self.assertEqual({r["evaluation_ms"] for r in rows}, {scale["evaluation_start_ms"], scale["evaluation_end_ms"]}) + self.assertTrue(all(r["scale"] == scale for r in rows)) + + def test_smoke_plan_drives_actual_generation(self): + """Plan dimensions override CLI defaults and survive in the data receipt.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + scale = plan("smoke", groups=1, members=1) + (root / "scale.json").write_text(json.dumps(scale)) + with patch("sys.argv", ["dataset.py", "--dataset", "synthetic", "--scale-plan", str(root / "scale.json"), + "--output", str(root / "data")]), patch("builtins.print"): + dataset.main() + meta = json.loads((root / "data/data-manifest.json").read_text()) + self.assertEqual(meta["samples"], scale["total_samples"]) + self.assertEqual(meta["series"], scale["total_series"]) + self.assertEqual(meta["provenance"]["scale_plan"], scale) + + def test_runner_rejects_undersized_data_and_shortened_repetition(self): + """A run cannot advertise a large scale while loading less or replaying less.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + scale = plan() + manifest = root / "queries.json" + manifest.write_text(json.dumps(query_manifest(scale))) + receipt = root / "receipt.json" + meta = {"provenance": {"scale_plan": scale}, "samples": 100, "series": 2} + receipt.write_text(json.dumps({"data": meta})) + args = argparse.Namespace(manifest=manifest, loaded_data=receipt, start_ms=None, end_ms=None) + with self.assertRaisesRegex(ValueError, "requested scale"): + accuracy_suite.run(args) + meta.update(samples=scale["total_samples"], series=scale["total_series"]) + receipt.write_text(json.dumps({"data": meta})) + args.end_ms = scale["evaluation_start_ms"] + with self.assertRaisesRegex(ValueError, "repetition interval"): + accuracy_suite.run(args) + + +if __name__ == "__main__": + unittest.main()