diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index 2abfd2ad..d814aa9c 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -9,27 +9,44 @@ value field in description order. For a table `.json` this writes, under OUT_CONFIG_DIR: -- /init.sql: the metric tables; -- /interval_s.sql: one statement per query that repeats every T s; +- /init.sql: the metric tables (baseline); +- /init_mv.sql: the metric tables and their materialized views + (baseline_mv); +- /interval_s_.sql: one statement per query that repeats every + T s, for the baseline (raw table) and baseline_mv (views) arms; - /asapquery.sql: each structured entry's `sql` (ASAPQuery's SQL, as sketch-bench's exporter writes it) with NOW() replaced by the floored eval time; -- _clickhouse.yaml: a baseline-only experiment_run_clickhouse.py config - with one query group per T, repeating every T seconds. +- _clickhouse.yaml, _clickhouse_mv.yaml: one + experiment_run_clickhouse.py config per arm, each loading data through its + own init SQL, with one query group per T, repeating every T seconds. + +baseline_mv keeps exact mergeable states, no sketches: one view per (metric, +value, statistic, T) holds per T-second bucket and group the state of sums, +counts, exact distinct counts, exact quantiles, each series' latest sample (instant vectors) or each +series' first/last sample, sample times and count (rate, increase). A query +at eval time E merges the W/T buckets of its window, plus for PromQL the +samples at exactly E - W (kept apart in each bucket's `edge` rows), so W must +be a multiple of T; instant vectors read [E - 1s, E], which is bucket-aligned +only when T = 1 s. The views live in VIEW_DATABASE. +A table with any query the views cannot serve gets no baseline_mv arm (no +init_mv.sql, MV SQL or _mv config) and a note on stderr naming the queries: +dropping just those queries would shift the arm's query indices against the +baseline SQL that labels both arms' results. Semantics are those of Prometheus 2.53 (the version installation/prometheus installs for bare-metal runs), whose windows are closed on both ends: range -vectors [d] at T read [T - d, T]; instant vectors read the latest sample of -each series in [T - 1s, T] (a lookback of one scrape interval, not +vectors [d] at E read [E - d, E]; instant vectors read the latest sample of +each series in [E - 1s, E] (a lookback of one scrape interval, not Prometheus's 5m default: the feeder writes every series every scrape, so the two agree unless a sample is missing, where Prometheus would return the series' last value and this SQL drops the series); quantiles use quantileExactInclusive; rate and increase reproduce extrapolatedRate (counters never reset). Structured queries involve no Prometheus: they read ASAPQuery's -left-open window (T - W, T], the window its SQL frontend and precompute +left-open window (E - W, E], the window its SQL frontend and precompute windows use, and pool all samples of a group: cardinality is uniqExact of -the value field, quantile is quantileExactInclusive. T is the client's -{{ end_time }} floored to a multiple of the query's interval. Unsupported +the value field, quantile is quantileExactInclusive. E is the client's +{{ end_time }} floored to a multiple of the query's interval T. Unsupported query shapes raise. Evaluation times: each query group sets `align_to_interval` and a @@ -46,6 +63,8 @@ import json import os import re +import shutil +import sys import warnings from dataclasses import dataclass, replace from typing import Callable, Dict, List, Optional, Tuple @@ -60,12 +79,23 @@ window_seconds, ) +sys.path.append( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +) +import constants # noqa: E402 + +BASELINE = constants.BASELINE_EXPERIMENT_NAME +BASELINE_MV = constants.BASELINE_MV_EXPERIMENT_NAME + OUT_CONFIG_DIR = "config/experiment_type/sketch_bench_clickhouse" CLICKHOUSE_URL = "http://localhost:8123" # Seconds after an eval time the client sends it, so data up to the eval # time has arrived. QUERY_TIME_OFFSET_SECS = 10 VALUE_COLUMN = "value" +# Every view lives here, so dropping it removes views any earlier workload +# attached to the metric tables; they would keep firing on inserts. +VIEW_DATABASE = "sketch_bench_views" # Instant vectors read the latest sample of each series in the last scrape. INSTANT_WINDOW_SECONDS = SCRAPE_INTERVAL_SECONDS @@ -359,8 +389,177 @@ def value_columns(desc: dict) -> List[str]: return list(desc["values"]) if desc["type"] == "records" else [VALUE_COLUMN] -def init_sql(metrics: dict) -> str: - statements = [] +@dataclass(frozen=True) +class View: + """A materialized view of `table` holding the mergeable state of one + statistic per bucket of `interval_secs` and `keys` group.""" + + name: str + table: str + stat: str # sum | count | cardinality | quantile | last | counter + interval_secs: int + keys: Tuple[str, ...] + levels: Tuple[str, ...] # quantile levels the view's state answers + value: str = VALUE_COLUMN + + +PER_SERIES_STATS = ("last", "counter") + + +def view_stat(q: Query) -> str: + if q.series_fn is None: + return q.agg + return "last" if q.series_fn == "last" else "counter" + + +def mv_servable(q: Query) -> bool: + """Whether the views' T-second buckets tile q's window.""" + return q.window_secs % q.interval_secs == 0 + + +def plan_views(queries: List[Query], metrics: dict) -> Dict[tuple, View]: + """One view per (table, value, statistic, interval), keyed by + view_key(). Pooled statistics keep the union of their queries' groupings + plus every label's child_of parents, so coarser groupings roll up; + per-series statistics keep every label.""" + grouped: Dict[tuple, List[Query]] = {} + for q in queries: + if not mv_servable(q): + raise ValueError( + f"window {q.window_secs}s is not a multiple of the " + f"{q.interval_secs}s buckets: {q}" + ) + grouped.setdefault(view_key(q), []).append(q) + views = {} + for key, qs in grouped.items(): + table, value, stat, interval = key + labels = metrics[table]["labels"] + parent = {label["name"]: label.get("child_of") for label in labels} + unknown = {p for p in parent.values() if p} - set(parent) + if unknown: + raise ValueError(f"{table!r}: child_of names unknown labels {unknown}") + keep = set(parent) if stat in PER_SERIES_STATS else set() + for name in (n for q in qs for n in q.grouping): + while name and name not in keep: + keep.add(name) + name = parent[name] + levels = {lv for q in qs for lv in (q.quantile, q.ratio_quantile) if lv} + views[key] = View( + name=f"{VIEW_DATABASE}.{table}_{value}_{stat}_{interval}s", + table=table, + stat=stat, + interval_secs=interval, + keys=tuple(label["name"] for label in labels if label["name"] in keep), + levels=tuple(sorted(levels, key=float)) if stat == "quantile" else (), + value=value, + ) + return views + + +def view_key(q: Query) -> tuple: + return (q.table, q.value, view_stat(q), q.interval_secs) + + +def view_states(view: View) -> List[Tuple[str, str]]: + """(column, -State expression) pairs the view stores.""" + v = view.value + if view.stat == "sum": + return [("sum_state", f"sumState({v})")] + if view.stat == "count": + return [("count_state", "countState()")] + if view.stat == "cardinality": + return [("cardinality_state", f"uniqExactState({v})")] + if view.stat == "quantile": + levels = ", ".join(view.levels) + return [("quantile_state", f"quantilesExactInclusiveState({levels})({v})")] + if view.stat == "last": + return [("last_state", f"argMaxState({v}, ts)")] + if view.stat == "counter": + return [ + ("first_state", f"argMinState({v}, ts)"), + ("last_state", f"argMaxState({v}, ts)"), + ("first_ts_state", "minState(ts)"), + ("last_ts_state", "maxState(ts)"), + ("n_state", "countState()"), + ] + raise ValueError(f"unknown statistic {view.stat!r}") + + +def view_ddl(view: View) -> str: + # Bucket b holds the samples in (b - T, b]; `edge` marks the one instant + # b. A closed window [E - W, E] is the buckets in (E - W, E] plus the edge + # of bucket E - W. + t = view.interval_secs + bucket = ( + f"toStartOfInterval(ts - INTERVAL 1 MILLISECOND, INTERVAL {t} SECOND) " + f"+ INTERVAL {t} SECOND" + ) + edge = f"toUnixTimestamp64Milli(ts) % {t * 1000} = 0" + keys = ", ".join(("bucket", "edge") + view.keys) + select = ", ".join( + (f"{bucket} AS bucket", f"{edge} AS edge") + + view.keys + + tuple(f"{expr} AS {col}" for col, expr in view_states(view)) + ) + return ( + f"CREATE MATERIALIZED VIEW {view.name} ENGINE = AggregatingMergeTree " + f"ORDER BY ({keys}) AS SELECT {select} FROM {view.table} GROUP BY {keys}" + ) + + +def mv_agg(view: View, stat: str, quantile: Optional[str]) -> str: + """-Merge of the view's state; a quantile picks its level from the + view's levels array.""" + if stat == "sum": + return "sumMerge(sum_state)" + if stat == "count": + return "countMerge(count_state)" + if stat == "cardinality": + return "uniqExactMerge(cardinality_state)" + if stat == "quantile": + index = view.levels.index(quantile) + 1 + levels = ", ".join(view.levels) + return f"quantilesExactInclusiveMerge({levels})(quantile_state)[{index}]" + if stat == "last": + return "argMaxMerge(last_state)" + if stat == "counter": + return ( + "argMinMerge(first_state) AS first_v, argMaxMerge(last_state) AS last_v, " + "toFloat64(minMerge(first_ts_state)) AS first_t, " + "toFloat64(maxMerge(last_ts_state)) AS last_t, countMerge(n_state) AS n" + ) + raise ValueError(f"unknown statistic {stat!r}") + + +def mv_sql(q: Query, views: Dict[tuple, View], labels: Dict[str, List[str]]) -> str: + """`q` over its view's buckets, merging their states.""" + view = views[view_key(q)] + return render( + q, + tuple(labels[q.table]), + view.name, + # Buckets b in (end - w, end] hold (end - w, end]; for a closed + # window, the edge row of bucket end - w holds its left boundary. + # `bucket` is a DateTime, so its bounds are too (see raw_sql). + lambda end, w: ( + f"bucket > toDateTime({end} - {w}) AND bucket <= toDateTime({end})" + if q.left_open + else f"bucket >= toDateTime({end} - {w}) AND bucket <= toDateTime({end}) " + f"AND (edge OR bucket > toDateTime({end} - {w}))" + ), + lambda stat, quantile: mv_agg(view, stat, quantile), + ) + + +def init_sql(metrics: dict, views: List[View], create_views: bool) -> str: + """The metric tables, and with `create_views` their views. A view only + aggregates rows inserted after it exists, so this must run before any + data arrives. VIEW_DATABASE is dropped either way, so no view left by an + earlier baseline_mv run, of this or any other workload, feeds on the + recreated tables.""" + statements = [f"DROP DATABASE IF EXISTS {VIEW_DATABASE}"] + if create_views: + statements.append(f"CREATE DATABASE {VIEW_DATABASE}") for table, desc in metrics.items(): names = [label["name"] for label in desc["labels"]] if not names: @@ -373,6 +572,8 @@ def init_sql(metrics: dict) -> str: f"CREATE TABLE {table} (ts DateTime64(3), {', '.join(columns)}) " f"ENGINE = MergeTree ORDER BY ({', '.join(names)}, ts)", ] + if create_views: + statements += [view_ddl(view) for view in views] return ";\n\n".join(statements) + ";\n" @@ -383,14 +584,16 @@ def metric_labels(metrics: dict) -> Dict[str, List[str]]: } -def write_statements(path: str, statements: List[str]) -> None: - with open(path, "w") as f: - f.write(";\n\n".join(statements) + ";\n") +def statements_text(statements: List[str]) -> str: + return ";\n\n".join(statements) + ";\n" def convert( table_path: str, data_file: str, starting_delay_secs: int, duration_secs: int -) -> str: +) -> List[str]: + """Writes the SQL files and one config per arm; returns the config paths. + Everything is rendered before anything is written, so a table that fails + leaves an earlier run's files intact.""" with open(table_path) as f: workloads = json.load(f)["workloads"] if len(workloads) != 1: @@ -400,31 +603,51 @@ def convert( labels = metric_labels(metrics) stem = os.path.splitext(os.path.basename(table_path))[0] sql_dir = os.path.join(OUT_CONFIG_DIR, stem) - os.makedirs(sql_dir, exist_ok=True) - - init_path = os.path.join(sql_dir, "init.sql") - with open(init_path, "w") as f: - f.write(init_sql(metrics)) + outputs: Dict[str, str] = {} # path -> content - by_interval: Dict[int, List[str]] = {} - asapquery: List[str] = [] - max_window = 0 - for entry in e2e["queries"]: - q = query_from_entry(entry, metrics, labels) + entries = [(e, query_from_entry(e, metrics, labels)) for e in e2e["queries"]] + for entry, q in entries: if q.interval_secs > duration_secs: raise ValueError( f"interval {q.interval_secs}s exceeds --duration-secs " f"{duration_secs}: {describe(entry)}" ) - max_window = max(max_window, q.window_secs) - by_interval.setdefault(q.interval_secs, []).append( - f"-- {describe(entry)}\n{raw_sql(q, labels)}" + unservable = [describe(e) for e, q in entries if not mv_servable(q)] + if unservable: + print( + f"{table_path}: no {BASELINE_MV} arm; its views need windows that " + f"are multiples of the interval: {'; '.join(unservable)}", + file=sys.stderr, + ) + modes, views = [BASELINE], {} + else: + modes = [BASELINE, BASELINE_MV] + views = plan_views([q for _, q in entries], metrics) + # Each arm loads data through its own init SQL, so baseline ingest never + # feeds the views. + init_paths = { + BASELINE: os.path.join(sql_dir, "init.sql"), + BASELINE_MV: os.path.join(sql_dir, "init_mv.sql"), + } + for mode in modes: + outputs[init_paths[mode]] = init_sql( + metrics, list(views.values()), mode == BASELINE_MV ) + + by_interval: Dict[int, Dict[str, List[str]]] = {} + asapquery: List[str] = [] + for entry, q in entries: + files = by_interval.setdefault(q.interval_secs, {m: [] for m in modes}) + files[BASELINE].append(f"-- {describe(entry)}\n{raw_sql(q, labels)}") + if BASELINE_MV in files: + sql = mv_sql(q, views, labels) + files[BASELINE_MV].append(f"-- {describe(entry)}\n{sql}") if "promql" not in entry: asapquery.append(f"-- {describe(entry)}\n{asapquery_sql(entry)}") if asapquery: # For the ASAPQuery arm; the runner does not read it. - write_statements(os.path.join(sql_dir, "asapquery.sql"), asapquery) + outputs[os.path.join(sql_dir, "asapquery.sql")] = statements_text(asapquery) + max_window = max((q.window_secs for _, q in entries), default=0) if starting_delay_secs < QUERY_TIME_OFFSET_SECS + max_window: # Same data for every arm, so arms stay comparable; absolutes are biased. warnings.warn( @@ -434,14 +657,16 @@ def convert( ) query_groups = [] - for interval, statements in sorted(by_interval.items()): + for interval, files in sorted(by_interval.items()): name = f"interval_{interval}s" - sql_path = os.path.join(sql_dir, f"{name}.sql") - write_statements(sql_path, statements) + sql_files = {} + for mode, statements in files.items(): + sql_files[mode] = os.path.join(sql_dir, f"{name}_{mode}.sql") + outputs[sql_files[mode]] = statements_text(statements) query_groups.append( { "name": name, - "sql_file": {"baseline": sql_path}, + "sql_file": sql_files, "repetition_delay_ms": interval * 1000, # Queries use only {{ end_time }}; the client still requires a value. "time_window_seconds": 0, @@ -454,26 +679,43 @@ def convert( } ) - out = { - "experiment": [{"mode": "baseline", "server": "clickhouse"}], - "servers": [{"name": "clickhouse", "url": CLICKHOUSE_URL}], - "dataset": { - "name": "custom", - "local_data_file": data_file, - # The runner's loader fills only this table; nothing fills the - # other metrics' tables yet. - "table": next(iter(labels)), - "init_sql_file": init_path, - "max_rows": 0, - }, - "query_groups": query_groups, - "data_ingestion_interval_ms": SCRAPE_INTERVAL_SECONDS * 1000, - } - out_path = os.path.join(OUT_CONFIG_DIR, f"{stem}_clickhouse.yaml") - with open(out_path, "w") as f: - f.write(f"# @package experiment_params\n# Generated from {table_path}\n") - yaml.safe_dump(out, f, sort_keys=False) - return out_path + out_paths = [] + for mode in modes: + suffix = "_mv" if mode == BASELINE_MV else "" + out = { + "experiment": [{"mode": mode, "server": "clickhouse"}], + "servers": [{"name": "clickhouse", "url": CLICKHOUSE_URL}], + "dataset": { + "name": "custom", + "local_data_file": data_file, + # The runner's loader fills only this table; nothing fills the + # other metrics' tables yet. + "table": next(iter(labels)), + "init_sql_file": init_paths[mode], + "max_rows": 0, + }, + # Both arms' SQL, so the saved config labels queries by baseline's text. + "query_groups": query_groups, + "data_ingestion_interval_ms": SCRAPE_INTERVAL_SECONDS * 1000, + } + out_path = os.path.join(OUT_CONFIG_DIR, f"{stem}_clickhouse{suffix}.yaml") + outputs[out_path] = ( + f"# @package experiment_params\n# Generated from {table_path}\n" + + yaml.safe_dump(out, sort_keys=False) + ) + out_paths.append(out_path) + + # Clear files an earlier run wrote, e.g. a baseline_mv arm or an interval + # this table no longer has. + shutil.rmtree(sql_dir, ignore_errors=True) + mv_config = os.path.join(OUT_CONFIG_DIR, f"{stem}_clickhouse_mv.yaml") + if os.path.exists(mv_config): + os.remove(mv_config) + os.makedirs(sql_dir) + for path, content in outputs.items(): + with open(path, "w") as f: + f.write(content) + return out_paths def main() -> None: @@ -487,11 +729,10 @@ def main() -> None: "--duration-secs", type=int, required=True, help="query phase length" ) args = parser.parse_args() - print( - convert( - args.table, args.data_file, args.starting_delay_secs, args.duration_secs - ) - ) + for path in convert( + args.table, args.data_file, args.starting_delay_secs, args.duration_secs + ): + print(path) if __name__ == "__main__": diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 7280945d..918d2138 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -93,7 +93,7 @@ import json import os import time -from typing import Callable, Optional +from typing import Callable, List, Optional from urllib.parse import urlparse import hydra @@ -297,12 +297,21 @@ def _run_query_workload( ) +def _ingest_mode(experiment_modes: List[str]) -> str: + """The mode whose output dir and client config the ingest monitor uses: + baseline when the run has it, else the first mode (e.g. a baseline_mv-only + run, whose init SQL attaches views that a baseline run must not pay for).""" + if constants.BASELINE_EXPERIMENT_NAME in experiment_modes: + return constants.BASELINE_EXPERIMENT_NAME + return experiment_modes[0] + + def _run_clickhouse_ingest_with_monitor( provider, node_offset: int, remote_monitor_service: RemoteMonitorService, - baseline_output_dir: str, - local_baseline_dir: str, + ingest_output_dir: str, + local_ingest_dir: str, controller_client_config: str, data_loader: ClickHouseDataLoaderService, dataset_name: str, @@ -314,10 +323,10 @@ def _run_clickhouse_ingest_with_monitor( monitor_interval_seconds: float = DEFAULT_MONITOR_INTERVAL_SECONDS, ) -> None: """Bulk-load netflow into ClickHouse while sampling ``clickhouse`` CPU/memory.""" - os.makedirs(local_baseline_dir, exist_ok=True) + os.makedirs(local_ingest_dir, exist_ok=True) provider.execute_command( node_idx=node_offset, - cmd=f"mkdir -p {baseline_output_dir}", + cmd=f"mkdir -p {ingest_output_dir}", cmd_dir="", nohup=False, popen=False, @@ -328,7 +337,7 @@ def _run_clickhouse_ingest_with_monitor( f"({monitor_interval_seconds}s samples)..." ) ingest_monitor_remote = os.path.join( - baseline_output_dir, + ingest_output_dir, "remote_monitor_output", CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE, ) @@ -342,7 +351,7 @@ def _run_clickhouse_ingest_with_monitor( ) remote_monitor_service.start_clickhouse_ingest_monitor( controller_client_config=controller_client_config, - experiment_output_dir=baseline_output_dir, + experiment_output_dir=ingest_output_dir, monitor_interval_seconds=monitor_interval_seconds, monitor_output_file=CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE, manual_mode=manual_remote_monitor, @@ -352,7 +361,7 @@ def _run_clickhouse_ingest_with_monitor( timeout=constants.REMOTE_MONITOR_START_TIMEOUT_SECONDS, polling_interval=constants.REMOTE_MONITOR_START_POLL_INTERVAL_SECONDS, execution_mode="ingest", - experiment_output_dir=baseline_output_dir, + experiment_output_dir=ingest_output_dir, ) time.sleep(constants.REMOTE_MONITOR_STARTUP_SETTLE_SECONDS) @@ -367,24 +376,24 @@ def _run_clickhouse_ingest_with_monitor( ) finally: print("Stopping ClickHouse ingest monitor...") - remote_monitor_service.signal_ingest_monitor_stop(baseline_output_dir) + remote_monitor_service.signal_ingest_monitor_stop(ingest_output_dir) try: remote_monitor_service.wait_for_remote_monitor_process_exit( polling_interval=constants.REMOTE_MONITOR_EXIT_POLL_INTERVAL_SECONDS, execution_mode="ingest", - experiment_output_dir=baseline_output_dir, + experiment_output_dir=ingest_output_dir, ) finally: - remote_monitor_service.cleanup_ingest_monitor_stop_file(baseline_output_dir) + remote_monitor_service.cleanup_ingest_monitor_stop_file(ingest_output_dir) sync.rsync_experiment_data( provider, - baseline_output_dir, - local_baseline_dir, + ingest_output_dir, + local_ingest_dir, node_offset=node_offset, ) ingest_monitor_path = os.path.join( - local_baseline_dir, + local_ingest_dir, "remote_monitor_output", CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE, ) @@ -394,14 +403,14 @@ def _run_clickhouse_ingest_with_monitor( if not ingest_data: raise RuntimeError( f"ClickHouse ingest monitor wrote empty/null data to " - f"{ingest_monitor_path}. Check baseline/remote_monitor.out and " - "baseline/remote_monitor_output/remote_monitor.log on the node." + f"{ingest_monitor_path}. Check /remote_monitor.out and " + "/remote_monitor_output/remote_monitor.log on the node." ) elif not manual_remote_monitor: raise RuntimeError( "ClickHouse ingest monitor output file was not created at " f"{ingest_monitor_path} (absent after stop timeout or write skip). " - "Check baseline/remote_monitor.out on the node." + "Check /remote_monitor.out on the node." ) print(f"ClickHouse ingest monitor output: {ingest_monitor_path}") @@ -530,25 +539,22 @@ def main(cfg: DictConfig) -> None: node_offset=node_offset, clickhouse_http_port=clickhouse_http_port, ) - baseline_client_config = os.path.join( + ingest_mode = _ingest_mode(experiment_modes) + ingest_client_config = os.path.join( experiment_root_output_dir, "controller_client_configs", - f"{constants.BASELINE_EXPERIMENT_NAME}.yaml", - ) - baseline_output_dir = os.path.join( - experiment_root_output_dir, constants.BASELINE_EXPERIMENT_NAME - ) - local_baseline_dir = os.path.join( - local_experiment_root_dir, constants.BASELINE_EXPERIMENT_NAME + f"{ingest_mode}.yaml", ) + ingest_output_dir = os.path.join(experiment_root_output_dir, ingest_mode) + local_ingest_dir = os.path.join(local_experiment_root_dir, ingest_mode) remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset) _run_clickhouse_ingest_with_monitor( provider=provider, node_offset=node_offset, remote_monitor_service=remote_monitor_service, - baseline_output_dir=baseline_output_dir, - local_baseline_dir=local_baseline_dir, - controller_client_config=baseline_client_config, + ingest_output_dir=ingest_output_dir, + local_ingest_dir=local_ingest_dir, + controller_client_config=ingest_client_config, data_loader=data_loader, dataset_name=dataset_name, remote_data_file=remote_data_file, diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index 6bbf4b60..e73370c3 100644 --- a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py +++ b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py @@ -7,6 +7,7 @@ from experiment_run_clickhouse import ( _deploy_prebuilt_clickhouse_controller_output, + _ingest_mode, _prepare_clickhouse_controller_output, _start_clickhouse_controller, ) @@ -81,5 +82,13 @@ def test_unset_prebuilt_path_generates_a_controller_output(self): deploy_prebuilt_output.assert_not_called() +class IngestModeTest(unittest.TestCase): + def test_baseline_when_present_else_first_mode(self): + self.assertEqual(_ingest_mode(["baseline", "sketchdb"]), "baseline") + self.assertEqual(_ingest_mode(["sketchdb", "baseline"]), "baseline") + # A baseline_mv-only run monitors ingest under its own mode. + self.assertEqual(_ingest_mode(["baseline_mv"]), "baseline_mv") + + if __name__ == "__main__": unittest.main() diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 55f11fd0..5f1ad016 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -6,7 +6,9 @@ CLICKHOUSE_BINARY) and is skipped otherwise. """ +import contextlib import datetime +import io import itertools import json import math @@ -170,9 +172,17 @@ def test_asapquery_sql_strips_trailing_semicolon(self): entry["sql"] += "; \n" self.assertEqual(sbw.asapquery_sql(entry), plain) + def test_unknown_child_of_raises(self): + metrics = { + "data": {"type": "gauge", "labels": [{"name": "job", "child_of": "dc"}]} + } + q = sbw.parse_promql("sum_over_time(data[1m])", 60, {"data": ["job"]}) + with self.assertRaises(ValueError): + sbw.plan_views([q], metrics) + def test_metric_without_labels_raises(self): with self.assertRaises(ValueError): - sbw.init_sql({"data": {"type": "gauge", "labels": []}}) + sbw.init_sql({"data": {"type": "gauge", "labels": []}}, [], False) def test_describe_is_one_line(self): entry = {"promql": "sum by (job) (\n data\n)", "interval_secs": 1} @@ -203,6 +213,50 @@ def mg_entry(capability: str, grouping: List[str], window_secs: int) -> dict: return entry +def snapshot(root: str) -> Dict[str, str]: + """Every file under `root`, by path, with its content.""" + out = {} + for d, _, names in os.walk(root): + for name in names: + with open(os.path.join(d, name)) as f: + out[os.path.join(d, name)] = f.read() + return out + + +class PlanViewsTest(unittest.TestCase): + def views(self, cases): + queries = [sbw.parse_promql(p, t, LABELS) for t, p in cases] + return sorted(sbw.plan_views(queries, METRICS).values(), key=lambda v: v.name) + + def test_classic_templates(self): + series = ("job", "label_0") + db = "sketch_bench_views." + self.assertEqual( + [(v.name, v.keys, v.levels) for v in self.views(CLASSIC)], + [ + (db + "data_total_value_counter_60s", series, ()), + (db + "data_value_last_1s", series, ()), + (db + "data_value_quantile_60s", series, ("0.5", "0.9", "0.99")), + (db + "data_value_sum_60s", series, ()), + ], + ) + + def test_keys_are_union_of_groupings_with_parents(self): + by_label_0 = (60, "sum by (label_0) (sum_over_time(data[1m]))") + by_job = (60, "sum by (job) (sum_over_time(data[1m]))") + self.assertEqual(self.views([by_job])[0].keys, ("job",)) + self.assertEqual(self.views([by_label_0])[0].keys, ("job", "label_0")) + + def test_window_not_multiple_of_interval_raises(self): + for case in [ + (60, "sum_over_time(data[90s])"), + # An instant vector reads [E - 1s, E]: bucket-aligned only when T = 1 s. + (60, "sum by (job) (data)"), + ]: + with self.subTest(case=case), self.assertRaises(ValueError): + self.views([case]) + + class FlooredEndDatetimeTest(unittest.TestCase): def test_renders_floored_utc_datetime(self): env = jinja2.Environment() @@ -246,9 +300,17 @@ def test_writes_config_grouped_by_interval(self): try: # 30 s < offset + the 15m windows: early windows predate data. with self.assertWarns(UserWarning): - out_path = sbw.convert(table, "data.json", 30, 600) - with open(out_path) as f: - cfg = yaml.safe_load(f) + out_paths = sbw.convert(table, "data.json", 30, 600) + cfgs = {} + for path in out_paths: + with open(path) as f: + cfg = yaml.safe_load(f) + (arm,) = cfg["experiment"] + with open(cfg["dataset"]["init_sql_file"]) as f: + cfgs[arm["mode"]] = (cfg, f.read()) + self.assertEqual(sorted(cfgs), ["baseline", "baseline_mv"]) + cfg, init_mv = cfgs["baseline_mv"] + _, init = cfgs["baseline"] groups = {g["name"]: g for g in cfg["query_groups"]} self.assertEqual(sorted(groups), ["interval_1s", "interval_60s"]) g60 = groups["interval_60s"] @@ -265,17 +327,28 @@ def test_writes_config_grouped_by_interval(self): self.assertEqual( groups["interval_1s"]["client_options"]["repetitions"], 600 ) - with open(g60["sql_file"]["baseline"]) as f: - statements = [s for s in f.read().split(";") if s.strip()] - self.assertEqual(len(statements), 9) - with open(cfg["dataset"]["init_sql_file"]) as f: - init = f.read() - self.assertIn("CREATE TABLE data_total (ts DateTime64(3)", init) + self.assertEqual(sorted(g60["sql_file"]), ["baseline", "baseline_mv"]) + for path in g60["sql_file"].values(): + with open(path) as f: + statements = [s for s in f.read().split(";") if s.strip()] + self.assertEqual(len(statements), 9) self.assertIn( "endpoint String, status String, latency Float64, " "user_id Float64)", init, ) + # Baseline ingest feeds no views, and drops any an earlier + # baseline_mv run left attached to the tables. + self.assertNotIn("CREATE MATERIALIZED VIEW", init) + self.assertIn("DROP DATABASE IF EXISTS sketch_bench_views", init) + # Views only see rows inserted after them: tables, then views. + self.assertLess( + init_mv.index("CREATE TABLE data_total (ts DateTime64(3)"), + init_mv.index( + "CREATE MATERIALIZED VIEW " + "sketch_bench_views.data_total_value_counter_60s" + ), + ) sql_dir = os.path.dirname(g60["sql_file"]["baseline"]) # The table's sql as written, with only NOW() substituted. end = "'" + sbw.floored_end_datetime(60) + "'" @@ -289,6 +362,93 @@ def test_writes_config_grouped_by_interval(self): finally: os.chdir(cwd) + def test_unservable_query_skips_mv_arm(self): + # An instant vector at E with T = 60 s reads [E - 1s, E], not whole + # buckets. + servable = { + "promql": "sum_over_time(data[1m])", + "interval_secs": 60, + "rqes": [], + } + instant = {"promql": "sum by (job) (data)", "interval_secs": 60, "rqes": []} + with tempfile.TemporaryDirectory() as tmp: + table = os.path.join(tmp, "instant.json") + cwd = os.getcwd() + os.chdir(tmp) + try: + # An earlier run of the same table that had a baseline_mv arm + # and a 1 s group; none of its files may survive. + with open(table, "w") as f: + e2e = { + "metrics": METRICS, + "queries": [ + servable, + { + **servable, + "interval_secs": 1, + "promql": "sum_over_time(data[1s])", + }, + ], + } + json.dump({"workloads": [{"e2e": e2e}]}, f) + with self.assertWarns(UserWarning): + sbw.convert(table, "data.json", 30, 600) + with open(table, "w") as f: + e2e = {"metrics": METRICS, "queries": [instant, servable]} + json.dump({"workloads": [{"e2e": e2e}]}, f) + with contextlib.redirect_stderr(io.StringIO()) as err: + (path,) = sbw.convert(table, "data.json", 30, 600) + self.assertIn("no baseline_mv arm", err.getvalue()) + self.assertIn("sum by (job) (data)", err.getvalue()) + with open(path) as f: + cfg = yaml.safe_load(f) + self.assertEqual( + cfg["experiment"], [{"mode": "baseline", "server": "clickhouse"}] + ) + (group,) = cfg["query_groups"] + self.assertEqual(list(group["sql_file"]), ["baseline"]) + with open(group["sql_file"]["baseline"]) as f: + self.assertEqual(f.read().count(";"), 2) + with open(cfg["dataset"]["init_sql_file"]) as f: + self.assertNotIn("MATERIALIZED VIEW", f.read()) + sql_dir = os.path.dirname(cfg["dataset"]["init_sql_file"]) + self.assertEqual( + sorted(os.listdir(sql_dir)), + ["init.sql", "interval_60s_baseline.sql"], + ) + self.assertEqual( + sorted(os.listdir(sbw.OUT_CONFIG_DIR)), + ["instant", "instant_clickhouse.yaml"], + ) + finally: + os.chdir(cwd) + + def test_failed_reconvert_keeps_earlier_files(self): + # Cleared before validating, a failing re-run left the earlier config + # pointing at deleted SQL files. + good = mg_entry("quantile", ["service"], 300) + bad = {**good, "sql": good["sql"].replace("NOW()", "now()")} + with tempfile.TemporaryDirectory() as tmp: + table = os.path.join(tmp, "t.json") + cwd = os.getcwd() + os.chdir(tmp) + try: + + def convert(queries): + with open(table, "w") as f: + e2e = {"metrics": METRICS, "queries": queries} + json.dump({"workloads": [{"e2e": e2e}]}, f) + return sbw.convert(table, "data.json", 310, 600) + + self.assertEqual(len(convert([good])), 2) + before = snapshot(sbw.OUT_CONFIG_DIR) + # Fails while rendering, after validation and the init SQL. + with self.assertRaises(ValueError): + convert([good, bad]) + self.assertEqual(snapshot(sbw.OUT_CONFIG_DIR), before) + finally: + os.chdir(cwd) + def test_checks_workloads_duration_and_delay(self): e2e = {"metrics": METRICS, "queries": [mg_entry("quantile", ["service"], 300)]} with tempfile.TemporaryDirectory() as tmp: @@ -584,10 +744,10 @@ def run_clickhouse(script: str, tmp: str) -> List[dict]: return [json.loads(line) for line in out.splitlines()] -def load_script(tmp: str, rows: Dict[str, List[tuple]]) -> str: +def load_script(tmp: str, rows: Dict[str, List[tuple]], views) -> str: """Init SQL, then each table's (ts ms, labels, values) rows inserted through it (and any views on it).""" - parts = [sbw.init_sql(METRICS)] + parts = [sbw.init_sql(METRICS, views, create_views=True)] for table, table_rows in rows.items(): names = LABELS[table] values = sbw.value_columns(METRICS[table]) @@ -652,7 +812,8 @@ def all_cases(gauge, counter, records) -> List[Tuple[str, "sbw.Query", Callable] class KeyPruningTest(unittest.TestCase): """ClickHouse 26.10 prunes `ts > ` on a DateTime64 key as `ts >= + 1`, so a left-open window lost the samples in its first - second. Small granules let the index prune them here too.""" + second. Small granules let the index prune them here too. The view's + DateTime `bucket` bounds are checked the same way.""" def test_samples_just_after_window_start_are_kept(self): end_s = DATA_START_MS // 1000 + 3600 @@ -673,61 +834,118 @@ def test_samples_just_after_window_start_are_kept(self): sbw.parse_promql("count_over_time(data[1m])", 60, LABELS), sum(start_ms <= t <= end_s * 1000 for t in times), ), + # The only query of its view and grouped by nothing, so the view + # has no label keys. + ( + sbw.query_from_entry(mg_entry("quantile", [], 60), METRICS, LABELS), + 1.0, + ), ] with tempfile.TemporaryDirectory() as tmp: - script = load_script(tmp, {"http": records, "data": series}).replace( - "ts);", "ts) SETTINGS index_granularity = 8;" + views = sbw.plan_views([q for q, _ in cases], METRICS) + rows = {"http": records, "data": series} + script = ( + load_script(tmp, rows, list(views.values())) + .replace("ts);", "ts) SETTINGS index_granularity = 8;") + .replace(") AS SELECT ", ") SETTINGS index_granularity = 1 AS SELECT ") ) - self.assertIn("index_granularity = 8", script) + self.assertEqual(script.count("index_granularity = 8"), len(METRICS)) + self.assertEqual(script.count("index_granularity = 1 "), len(views)) statements = [ - (str(i), sbw.raw_sql(q, LABELS).replace(sbw.END, str(end_s))) + (f"{mode}/{i}", sql.replace(sbw.END, str(end_s))) for i, (q, _) in enumerate(cases) + for mode, sql in [ + ("raw", sbw.raw_sql(q, LABELS)), + ("mv", sbw.mv_sql(q, views, LABELS)), + ] ] - rows = run_clickhouse(script + query_script(statements), tmp) - got = {row["qid"]: row["value"] for row in rows} - self.assertEqual(got, {str(i): want for i, (_, want) in enumerate(cases)}) + out = run_clickhouse(script + query_script(statements), tmp) + got = {row["qid"]: row["value"] for row in out} + want = { + f"{mode}/{i}": n for i, (_, n) in enumerate(cases) for mode in ("raw", "mv") + } + self.assertEqual(got, want) + + +@unittest.skipUnless(CLICKHOUSE, "clickhouse binary not found") +class StaleViewsTest(unittest.TestCase): + def test_init_drops_other_workloads_views(self): + """A view stays attached to a dropped and recreated table, so one + left by another workload's baseline_mv run would keep firing.""" + q = sbw.parse_promql("sum_over_time(data[1m])", 60, LABELS) + other = list(sbw.plan_views([q], METRICS).values()) + with tempfile.TemporaryDirectory() as tmp: + script = ( + sbw.init_sql(METRICS, other, True) + # This workload has no views of its own. + + load_script(tmp, {"data": [(DATA_START_MS, ("j", "l"), (1.0,))]}, []) + + "SELECT count() AS n FROM system.tables " + f"WHERE database = '{sbw.VIEW_DATABASE}' FORMAT JSONEachRow;\n" + ) + self.assertEqual(run_clickhouse(script, tmp), [{"n": 0}]) @unittest.skipUnless(CLICKHOUSE, "clickhouse binary not found") class ClickHouseSemanticsTest(unittest.TestCase): - def test_raw_sql_matches_prometheus_reference(self): + """Every CASES and MULTIGROUP query at every eval offset, over the raw + table (baseline) and over its view (baseline_mv), in one clickhouse-local + run.""" + + @classmethod + def setUpClass(cls): gauge, counter = make_dataset() records = make_records() - cases = all_cases(gauge, counter, records) + cls.cases = all_cases(gauge, counter, records) + views = sbw.plan_views([q for _, q, _ in cls.cases], METRICS) statements, groupings = [], {} - for i, (_, q, _) in enumerate(cases): + for i, (_, q, _) in enumerate(cls.cases): for offset in EVAL_OFFSETS_SECS[q.interval_secs]: - end = DATA_START_MS // 1000 + offset - qid = f"{i}@{offset}" - sql = sbw.raw_sql(q, LABELS).replace(sbw.END, str(end)) - statements.append((qid, sql)) - groupings[qid] = q.grouping + end = str(DATA_START_MS // 1000 + offset) + for mode, sql in [ + ("raw", sbw.raw_sql(q, LABELS)), + ("mv", sbw.mv_sql(q, views, LABELS)), + ]: + qid = f"{mode}/{i}@{offset}" + statements.append((qid, sql.replace(sbw.END, end))) + groupings[qid] = q.grouping rows = { "data": series_rows(gauge), "data_total": series_rows(counter), "http": records, } with tempfile.TemporaryDirectory() as tmp: - script = load_script(tmp, rows) - got = rows_by_id( + script = load_script(tmp, rows, list(views.values())) + cls.got = rows_by_id( run_clickhouse(script + query_script(statements), tmp), groupings ) + def test_raw_sql_matches_prometheus_reference(self): BRANCHES_HIT.clear() - nonempty = 0 - for i, (name, q, ref) in enumerate(cases): + nonempty = total = 0 + for i, (name, q, ref) in enumerate(self.cases): for offset in EVAL_OFFSETS_SECS[q.interval_secs]: with self.subTest(query=name, offset=offset): want = ref(DATA_START_MS + offset * 1000) - nonempty += bool(want) - assert_vectors_close(self, got[f"{i}@{offset}"], want, 1e-9) + nonempty, total = nonempty + bool(want), total + 1 + got = self.got[f"raw/{i}@{offset}"] + assert_vectors_close(self, got, want, 1e-9) self.assertEqual( BRANCHES_HIT, {"start_avg", "start_boundary", "zero_clamp", "end_avg", "end_boundary"}, ) # Both empty and non-empty windows are compared. - self.assertGreater(nonempty, len(statements) // 2) - self.assertLess(nonempty, len(statements)) + self.assertGreater(nonempty, total // 2) + self.assertLess(nonempty, total) + + def test_mv_sql_matches_raw_sql(self): + for i, (name, q, _) in enumerate(self.cases): + # Only sums are re-associated (per bucket, then across buckets and + # groups); every other statistic merges to the identical value. + rel = 1e-9 if q.agg == "sum" else 0.0 + for offset in EVAL_OFFSETS_SECS[q.interval_secs]: + with self.subTest(query=name, offset=offset): + raw = self.got[f"raw/{i}@{offset}"] + assert_vectors_close(self, self.got[f"mv/{i}@{offset}"], raw, rel) if __name__ == "__main__":