feat(asap-tools): run generated2 PromQL workloads on ClickHouse

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()

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 <jsonl.gz>
+"""
+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()

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);

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"):