Skip to content

Commit dfa8022

Browse files
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 <noreply@anthropic.com>
1 parent 9d6727f commit dfa8022

2 files changed

Lines changed: 218 additions & 1 deletion

File tree

Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,206 @@
1+
#!/usr/bin/env python3
2+
"""Translate generated2 PromQL workloads into ClickHouse experiment configs.
3+
4+
For each generated2 workload yaml (Prometheus variants only), writes a
5+
baseline-only ClickHouse experiment config plus one templated SQL file per
6+
query group under config/experiment_type/generated2_clickhouse/ (gitignored,
7+
like its generated2 inputs). Query groups keep their
8+
original delays, repetitions and query_time_offset; the client fills in
9+
{{ end_time }} (= T, the PromQL evaluation time) per repetition.
10+
11+
Semantics follow Prometheus 2.53 with these windows:
12+
- instant vector at T: samples in (T - SCRAPE_INTERVAL_SECONDS, T], i.e. the
13+
one scrape at T;
14+
- range vector [d] at T: samples in (T - d, T].
15+
rate/increase reproduce extrapolatedRate. Only the query shapes that appear
16+
in generated2 are supported; anything else raises.
17+
18+
Run from asap-tools/experiments:
19+
python3 datasets/fake_metric/generate_workloads.py \
20+
--src-dir config/experiment_type/generated2 --data-file <jsonl.gz>
21+
"""
22+
import argparse
23+
import glob
24+
import os
25+
import re
26+
from typing import List
27+
28+
import yaml
29+
30+
from generate_data import DATA_START
31+
32+
SCRAPE_INTERVAL_SECONDS = 1 # config/config.yaml prometheus.scrape_interval
33+
OUT_CONFIG_DIR = "config/experiment_type/generated2_clickhouse"
34+
INIT_SQL = "datasets/fake_metric/init.sql"
35+
END = "{{ end_time }}"
36+
37+
DURATION_UNITS = {"s": 1, "m": 60, "h": 3600}
38+
RANGE = r"(\w+)\[(\d+)([smh])\]"
39+
SHAPES = [
40+
("instant_agg", re.compile(r"^(count|sum) by \((\w+)\) \((\w+)\)$")),
41+
("instant_quantile", re.compile(r"^quantile by \((\w+)\) \(([\d.]+), (\w+)\)$")),
42+
(
43+
"agg_over_time",
44+
re.compile(rf"^sum by \((\w+)\) \((sum|count)_over_time\({RANGE}\)\)$"),
45+
),
46+
("over_time", re.compile(rf"^(sum|count)_over_time\({RANGE}\)$")),
47+
("quantile_over_time", re.compile(rf"^quantile_over_time\(([\d.]+), {RANGE}\)$")),
48+
("rate", re.compile(rf"^(rate|increase)\({RANGE}\)$")),
49+
]
50+
AGG_SQL = {"sum": "sum(value)", "count": "count()"}
51+
52+
53+
def window_seconds(amount: str, unit: str) -> int:
54+
return int(amount) * DURATION_UNITS[unit]
55+
56+
57+
def select(cols: str, agg: str, table: str, window: int) -> str:
58+
return (
59+
f"SELECT {cols}, {agg} AS value FROM {table} "
60+
f"WHERE ts > {END} - {window} AND ts <= {END} GROUP BY {cols}"
61+
)
62+
63+
64+
def extrapolated_rate(cols: str, table: str, window: int, is_rate: bool) -> str:
65+
"""Prometheus 2.53 extrapolatedRate, isCounter=true."""
66+
# ponytail: no counter-reset correction, fake_exporter counters only grow
67+
samples = (
68+
f"SELECT {cols}, argMin(value, ts) AS first_v, argMax(value, ts) AS last_v, "
69+
f"toFloat64(toUnixTimestamp(min(ts))) AS first_t, "
70+
f"toFloat64(toUnixTimestamp(max(ts))) AS last_t, count() AS n "
71+
f"FROM {table} WHERE ts > {END} - {window} AND ts <= {END} "
72+
f"GROUP BY {cols} HAVING n >= 2"
73+
)
74+
durations = (
75+
f"SELECT {cols}, last_v - first_v AS delta, first_v, "
76+
f"last_t - first_t AS sampled, sampled / (n - 1) AS avg_step, "
77+
f"if(first_t - ({END} - {window}) >= 1.1 * avg_step, avg_step / 2, "
78+
f"first_t - ({END} - {window})) AS to_start_clamped, "
79+
f"if(delta > 0 AND first_v >= 0, "
80+
f"least(to_start_clamped, sampled * first_v / delta), to_start_clamped) AS to_start, "
81+
f"if({END} - last_t >= 1.1 * avg_step, avg_step / 2, {END} - last_t) AS to_end "
82+
f"FROM ({samples})"
83+
)
84+
per_second = f" / {window}" if is_rate else ""
85+
return (
86+
f"SELECT {cols}, delta * (sampled + to_start + to_end) / sampled{per_second} "
87+
f"AS value FROM ({durations})"
88+
)
89+
90+
91+
def translate(promql: str, metric: str, series_cols: str) -> str:
92+
query = " ".join(promql.split())
93+
for shape, pattern in SHAPES:
94+
m = pattern.match(query)
95+
if not m:
96+
continue
97+
g = m.groups()
98+
if shape == "instant_agg":
99+
fn, by, name = g
100+
agg, cols, window = AGG_SQL[fn], by, SCRAPE_INTERVAL_SECONDS
101+
elif shape == "instant_quantile":
102+
by, q, name = g
103+
agg, cols, window = (
104+
f"quantileExactInclusive({q})(value)",
105+
by,
106+
SCRAPE_INTERVAL_SECONDS,
107+
)
108+
elif shape == "agg_over_time":
109+
by, fn, name, amount, unit = g
110+
agg, cols, window = AGG_SQL[fn], by, window_seconds(amount, unit)
111+
elif shape == "over_time":
112+
fn, name, amount, unit = g
113+
agg, cols, window = AGG_SQL[fn], series_cols, window_seconds(amount, unit)
114+
elif shape == "quantile_over_time":
115+
q, name, amount, unit = g
116+
agg, cols, window = (
117+
f"quantileExactInclusive({q})(value)",
118+
series_cols,
119+
window_seconds(amount, unit),
120+
)
121+
else:
122+
fn, name, amount, unit = g
123+
if name != metric:
124+
raise ValueError(f"unexpected metric {name!r} in {promql!r}")
125+
return extrapolated_rate(
126+
series_cols, metric, window_seconds(amount, unit), fn == "rate"
127+
)
128+
if name != metric:
129+
raise ValueError(f"unexpected metric {name!r} in {promql!r}")
130+
return select(cols, agg, metric, window)
131+
raise ValueError(f"unsupported PromQL shape: {promql!r}")
132+
133+
134+
def convert(src_path: str, data_file: str) -> str:
135+
with open(src_path) as f:
136+
src = yaml.safe_load(f)
137+
(metric_cfg,) = src["metrics"]
138+
metric = metric_cfg["metric"]
139+
series_cols = ", ".join(metric_cfg["labels"])
140+
stem = os.path.splitext(os.path.basename(src_path))[0]
141+
sql_dir = os.path.join(OUT_CONFIG_DIR, stem)
142+
os.makedirs(sql_dir, exist_ok=True)
143+
144+
query_groups = []
145+
for group in src["query_groups"]:
146+
sql_path = os.path.join(sql_dir, f"group_{group['id']}.sql")
147+
statements: List[str] = [
148+
f"-- {q}\n{translate(q, metric, series_cols)}" for q in group["queries"]
149+
]
150+
with open(sql_path, "w") as f:
151+
f.write(";\n\n".join(statements) + ";\n")
152+
query_groups.append(
153+
{
154+
"name": f"group_{group['id']}",
155+
"sql_file": {"baseline": sql_path},
156+
"repetition_delay_ms": group["repetition_delay_ms"],
157+
# Queries use only {{ end_time }}; the client still requires a value.
158+
"time_window_seconds": 0,
159+
"client_options": {
160+
**group["client_options"],
161+
"simulated_start_time": DATA_START,
162+
},
163+
"controller_options": group["controller_options"],
164+
}
165+
)
166+
167+
out = {
168+
"experiment": [{"mode": "baseline", "server": "clickhouse"}],
169+
"servers": [{"name": "clickhouse", "url": "http://localhost:8123"}],
170+
"dataset": {
171+
"name": "custom",
172+
"local_data_file": data_file,
173+
"table": metric,
174+
"init_sql_file": INIT_SQL,
175+
"max_rows": 0,
176+
},
177+
"query_groups": query_groups,
178+
"data_ingestion_interval_ms": SCRAPE_INTERVAL_SECONDS * 1000,
179+
}
180+
out_path = os.path.join(OUT_CONFIG_DIR, f"{stem}_clickhouse.yaml")
181+
with open(out_path, "w") as f:
182+
f.write(f"# @package experiment_params\n# Generated from {src_path}\n")
183+
yaml.safe_dump(out, f, sort_keys=False)
184+
return out_path
185+
186+
187+
def main() -> None:
188+
parser = argparse.ArgumentParser(description=__doc__)
189+
parser.add_argument("--src-dir", required=True, help="generated2 config dir")
190+
parser.add_argument(
191+
"--data-file",
192+
required=True,
193+
help="generate_data.py output; path as seen from asap-tools/experiments",
194+
)
195+
args = parser.parse_args()
196+
sources = sorted(
197+
p
198+
for p in glob.glob(os.path.join(args.src_dir, "*.yaml"))
199+
if not p.endswith("_victoriametrics.yaml")
200+
)
201+
for src in sources:
202+
print(convert(src, args.data_file))
203+
204+
205+
if __name__ == "__main__":
206+
main()

‎asap-tools/queriers/prometheus-client/main_prometheus_client.py‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -316,8 +316,19 @@ def handle_query_group(
316316
for server_name in servers
317317
}
318318

319+
simulated_start_time = getattr(query_group, "simulated_start_time", None)
320+
319321
for repetition_idx in range(query_group.repetitions):
320-
current_time = datetime.datetime.now()
322+
if simulated_start_time is None:
323+
current_time = datetime.datetime.now()
324+
else:
325+
# Pre-loaded data has fixed timestamps, so query time follows a
326+
# clock anchored at the data start instead of the wall clock.
327+
current_time = datetime.datetime.fromtimestamp(
328+
simulated_start_time
329+
+ query_group.starting_delay
330+
+ repetition_idx * query_group.repetition_delay_ms / 1000
331+
)
321332
logger.debug("Current unix time: {}", int(current_time.timestamp()))
322333

323334
if hasattr(query_group, "query_time_offset"):

0 commit comments

Comments
 (0)