From 9d6727fb21affa347f53defb3752adf2d5a8631a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 13:09:48 -0400 Subject: [PATCH 1/2] feat(experiments): fake_exporter-equivalent dataset for ClickHouse generated2 workloads Co-Authored-By: Claude Opus 5.5 --- .../datasets/fake_metric/generate_data.py | 84 +++++++++++++++++++ .../experiments/datasets/fake_metric/init.sql | 14 ++++ 2 files changed, 98 insertions(+) create mode 100644 asap-tools/experiments/datasets/fake_metric/generate_data.py create mode 100644 asap-tools/experiments/datasets/fake_metric/init.sql diff --git a/asap-tools/experiments/datasets/fake_metric/generate_data.py b/asap-tools/experiments/datasets/fake_metric/generate_data.py new file mode 100644 index 00000000..fbbaae6d --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/generate_data.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Generate fake_exporter-equivalent data for the ClickHouse generated2 workloads. + +Mirrors the generated2 Prometheus setup: one fake_exporter per instance +(zipf, valuescale 10000, 3 labels x 20 values, counter), scraped every 1s. +Values match fake_exporter in distribution, not sample-for-sample: +- each scrape draws one bounded Zipf(n=VALUE_SCALE, s=ZIPF_ALPHA) sample per + series and adds it to the instance's single shared counter, in series order + (fake_exporter's counter_state is per process, not per series); +- every series of every instance is stamped with the same 1s-aligned tick. + +Ticks are DATA_START + t for t in [0, duration). Writes gzipped JSON lines +with columns matching init.sql. +""" +import argparse +import gzip +import itertools +from typing import List + +import numpy as np + +DATA_START = 1767225600 # 2026-01-01 00:00:00 UTC; the SQL translator anchors T here +START_PORT = 50000 +JOB = "fake_exporter" +NUM_LABELS = 3 +NUM_VALUES_PER_LABEL = 20 +VALUE_SCALE = 10000 +ZIPF_ALPHA = 1.01 +# Last query in generated2 fires at T = starting_delay 70 + 1149 reps - offset 10. +DEFAULT_DURATION_SECONDS = 1210 + + +def zipf_cdf(n: int, alpha: float) -> np.ndarray: + weights = np.arange(1, n + 1, dtype=np.float64) ** -alpha + cdf = np.cumsum(weights) / weights.sum() + cdf[-1] = 1.0 # float rounding must not let searchsorted step past n + return cdf + + +def series_prefixes(instance: str) -> List[str]: + combos = itertools.product(range(NUM_VALUES_PER_LABEL), repeat=NUM_LABELS) + return [ + f'{{"instance":"{instance}","job":"{JOB}",' + + ",".join(f'"label_{i}":"value_{i}_value_{v}"' for i, v in enumerate(combo)) + + "," + for combo in combos + ] + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out", required=True, help="output .jsonl.gz path") + parser.add_argument("--num-instances", type=int, default=10) + parser.add_argument("--duration", type=int, default=DEFAULT_DURATION_SECONDS) + parser.add_argument("--seed", type=int, default=0) + args = parser.parse_args() + + rng = np.random.default_rng(args.seed) + cdf = zipf_cdf(VALUE_SCALE, ZIPF_ALPHA) + prefixes = [ + series_prefixes(f"localhost:{START_PORT + i}") + for i in range(args.num_instances) + ] + num_series = len(prefixes[0]) + counters = np.zeros(args.num_instances) + + # ponytail: gzip level 1, the loader re-parses every line so CPU here is not the bottleneck + with gzip.open(args.out, "wt", compresslevel=1) as f: + for t in range(args.duration): + ts = DATA_START + t + samples = ( + np.searchsorted(cdf, rng.random((args.num_instances, num_series))) + 1.0 + ) + values = counters[:, None] + np.cumsum(samples, axis=1) + counters = values[:, -1] + f.writelines( + f'{p}"ts":{ts},"value":{v!r}}}\n' + for inst_prefixes, inst_values in zip(prefixes, values.tolist()) + for p, v in zip(inst_prefixes, inst_values) + ) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/experiments/datasets/fake_metric/init.sql b/asap-tools/experiments/datasets/fake_metric/init.sql new file mode 100644 index 00000000..ea6b7b1b --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/init.sql @@ -0,0 +1,14 @@ +DROP TABLE IF EXISTS fake_metric_total; + +CREATE TABLE fake_metric_total +( + ts DateTime, + instance LowCardinality(String), + job LowCardinality(String), + label_0 LowCardinality(String), + label_1 LowCardinality(String), + label_2 LowCardinality(String), + value Float64 +) +ENGINE = MergeTree +ORDER BY (instance, label_0, label_1, label_2, ts); From dfa8022c1ca09b0d28cf25764837c30c4a0b23e6 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 13:17:21 -0400 Subject: [PATCH 2/2] feat(experiments): translate generated2 PromQL workloads to ClickHouse SQL configs Adds a client_options.simulated_start_time so templated SQL can be evaluated against bulk-loaded data with fixed timestamps. Co-Authored-By: Claude Opus 5.5 --- .../fake_metric/generate_workloads.py | 206 ++++++++++++++++++ .../main_prometheus_client.py | 13 +- 2 files changed, 218 insertions(+), 1 deletion(-) create mode 100644 asap-tools/experiments/datasets/fake_metric/generate_workloads.py diff --git a/asap-tools/experiments/datasets/fake_metric/generate_workloads.py b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py new file mode 100644 index 00000000..a68bb8b7 --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +"""Translate generated2 PromQL workloads into ClickHouse experiment configs. + +For each generated2 workload yaml (Prometheus variants only), writes a +baseline-only ClickHouse experiment config plus one templated SQL file per +query group under config/experiment_type/generated2_clickhouse/ (gitignored, +like its generated2 inputs). Query groups keep their +original delays, repetitions and query_time_offset; the client fills in +{{ end_time }} (= T, the PromQL evaluation time) per repetition. + +Semantics follow Prometheus 2.53 with these windows: +- instant vector at T: samples in (T - SCRAPE_INTERVAL_SECONDS, T], i.e. the + one scrape at T; +- range vector [d] at T: samples in (T - d, T]. +rate/increase reproduce extrapolatedRate. Only the query shapes that appear +in generated2 are supported; anything else raises. + +Run from asap-tools/experiments: + python3 datasets/fake_metric/generate_workloads.py \ + --src-dir config/experiment_type/generated2 --data-file +""" +import argparse +import glob +import os +import re +from typing import List + +import yaml + +from generate_data import DATA_START + +SCRAPE_INTERVAL_SECONDS = 1 # config/config.yaml prometheus.scrape_interval +OUT_CONFIG_DIR = "config/experiment_type/generated2_clickhouse" +INIT_SQL = "datasets/fake_metric/init.sql" +END = "{{ end_time }}" + +DURATION_UNITS = {"s": 1, "m": 60, "h": 3600} +RANGE = r"(\w+)\[(\d+)([smh])\]" +SHAPES = [ + ("instant_agg", re.compile(r"^(count|sum) by \((\w+)\) \((\w+)\)$")), + ("instant_quantile", re.compile(r"^quantile by \((\w+)\) \(([\d.]+), (\w+)\)$")), + ( + "agg_over_time", + re.compile(rf"^sum by \((\w+)\) \((sum|count)_over_time\({RANGE}\)\)$"), + ), + ("over_time", re.compile(rf"^(sum|count)_over_time\({RANGE}\)$")), + ("quantile_over_time", re.compile(rf"^quantile_over_time\(([\d.]+), {RANGE}\)$")), + ("rate", re.compile(rf"^(rate|increase)\({RANGE}\)$")), +] +AGG_SQL = {"sum": "sum(value)", "count": "count()"} + + +def window_seconds(amount: str, unit: str) -> int: + return int(amount) * DURATION_UNITS[unit] + + +def select(cols: str, agg: str, table: str, window: int) -> str: + return ( + f"SELECT {cols}, {agg} AS value FROM {table} " + f"WHERE ts > {END} - {window} AND ts <= {END} GROUP BY {cols}" + ) + + +def extrapolated_rate(cols: str, table: str, window: int, is_rate: bool) -> str: + """Prometheus 2.53 extrapolatedRate, isCounter=true.""" + # ponytail: no counter-reset correction, fake_exporter counters only grow + samples = ( + f"SELECT {cols}, argMin(value, ts) AS first_v, argMax(value, ts) AS last_v, " + f"toFloat64(toUnixTimestamp(min(ts))) AS first_t, " + f"toFloat64(toUnixTimestamp(max(ts))) AS last_t, count() AS n " + f"FROM {table} WHERE ts > {END} - {window} AND ts <= {END} " + f"GROUP BY {cols} HAVING n >= 2" + ) + durations = ( + f"SELECT {cols}, last_v - first_v AS delta, first_v, " + f"last_t - first_t AS sampled, sampled / (n - 1) AS avg_step, " + f"if(first_t - ({END} - {window}) >= 1.1 * avg_step, avg_step / 2, " + f"first_t - ({END} - {window})) AS to_start_clamped, " + f"if(delta > 0 AND first_v >= 0, " + f"least(to_start_clamped, sampled * first_v / delta), to_start_clamped) AS to_start, " + f"if({END} - last_t >= 1.1 * avg_step, avg_step / 2, {END} - last_t) AS to_end " + f"FROM ({samples})" + ) + per_second = f" / {window}" if is_rate else "" + return ( + f"SELECT {cols}, delta * (sampled + to_start + to_end) / sampled{per_second} " + f"AS value FROM ({durations})" + ) + + +def translate(promql: str, metric: str, series_cols: str) -> str: + query = " ".join(promql.split()) + for shape, pattern in SHAPES: + m = pattern.match(query) + if not m: + continue + g = m.groups() + if shape == "instant_agg": + fn, by, name = g + agg, cols, window = AGG_SQL[fn], by, SCRAPE_INTERVAL_SECONDS + elif shape == "instant_quantile": + by, q, name = g + agg, cols, window = ( + f"quantileExactInclusive({q})(value)", + by, + SCRAPE_INTERVAL_SECONDS, + ) + elif shape == "agg_over_time": + by, fn, name, amount, unit = g + agg, cols, window = AGG_SQL[fn], by, window_seconds(amount, unit) + elif shape == "over_time": + fn, name, amount, unit = g + agg, cols, window = AGG_SQL[fn], series_cols, window_seconds(amount, unit) + elif shape == "quantile_over_time": + q, name, amount, unit = g + agg, cols, window = ( + f"quantileExactInclusive({q})(value)", + series_cols, + window_seconds(amount, unit), + ) + else: + fn, name, amount, unit = g + if name != metric: + raise ValueError(f"unexpected metric {name!r} in {promql!r}") + return extrapolated_rate( + series_cols, metric, window_seconds(amount, unit), fn == "rate" + ) + if name != metric: + raise ValueError(f"unexpected metric {name!r} in {promql!r}") + return select(cols, agg, metric, window) + raise ValueError(f"unsupported PromQL shape: {promql!r}") + + +def convert(src_path: str, data_file: str) -> str: + with open(src_path) as f: + src = yaml.safe_load(f) + (metric_cfg,) = src["metrics"] + metric = metric_cfg["metric"] + series_cols = ", ".join(metric_cfg["labels"]) + stem = os.path.splitext(os.path.basename(src_path))[0] + sql_dir = os.path.join(OUT_CONFIG_DIR, stem) + os.makedirs(sql_dir, exist_ok=True) + + query_groups = [] + for group in src["query_groups"]: + sql_path = os.path.join(sql_dir, f"group_{group['id']}.sql") + statements: List[str] = [ + f"-- {q}\n{translate(q, metric, series_cols)}" for q in group["queries"] + ] + with open(sql_path, "w") as f: + f.write(";\n\n".join(statements) + ";\n") + query_groups.append( + { + "name": f"group_{group['id']}", + "sql_file": {"baseline": sql_path}, + "repetition_delay_ms": group["repetition_delay_ms"], + # Queries use only {{ end_time }}; the client still requires a value. + "time_window_seconds": 0, + "client_options": { + **group["client_options"], + "simulated_start_time": DATA_START, + }, + "controller_options": group["controller_options"], + } + ) + + out = { + "experiment": [{"mode": "baseline", "server": "clickhouse"}], + "servers": [{"name": "clickhouse", "url": "http://localhost:8123"}], + "dataset": { + "name": "custom", + "local_data_file": data_file, + "table": metric, + "init_sql_file": INIT_SQL, + "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 {src_path}\n") + yaml.safe_dump(out, f, sort_keys=False) + return out_path + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--src-dir", required=True, help="generated2 config dir") + parser.add_argument( + "--data-file", + required=True, + help="generate_data.py output; path as seen from asap-tools/experiments", + ) + args = parser.parse_args() + sources = sorted( + p + for p in glob.glob(os.path.join(args.src_dir, "*.yaml")) + if not p.endswith("_victoriametrics.yaml") + ) + for src in sources: + print(convert(src, args.data_file)) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/queriers/prometheus-client/main_prometheus_client.py b/asap-tools/queriers/prometheus-client/main_prometheus_client.py index 104e4d68..84be87f4 100644 --- a/asap-tools/queriers/prometheus-client/main_prometheus_client.py +++ b/asap-tools/queriers/prometheus-client/main_prometheus_client.py @@ -316,8 +316,19 @@ def handle_query_group( for server_name in servers } + simulated_start_time = getattr(query_group, "simulated_start_time", None) + for repetition_idx in range(query_group.repetitions): - current_time = datetime.datetime.now() + if simulated_start_time is None: + current_time = datetime.datetime.now() + else: + # Pre-loaded data has fixed timestamps, so query time follows a + # clock anchored at the data start instead of the wall clock. + current_time = datetime.datetime.fromtimestamp( + simulated_start_time + + query_group.starting_delay + + repetition_idx * query_group.repetition_delay_ms / 1000 + ) logger.debug("Current unix time: {}", int(current_time.timestamp())) if hasattr(query_group, "query_time_offset"):