Skip to content

Commit 37f9732

Browse files
bug fixes
1 parent bc26321 commit 37f9732

5 files changed

Lines changed: 142 additions & 23 deletions

File tree

‎asap-query-engine/src/drivers/query/servers/http.rs‎

Lines changed: 18 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -324,7 +324,7 @@ async fn handle_instant_query_post(
324324
.and_then(|v| v.to_str().ok())
325325
.unwrap_or("");
326326

327-
debug!("Content-Type: {}", content_type);
327+
debug!("POST content-type: '{}'", content_type);
328328

329329
let parsed_request = if content_type.contains("application/json") {
330330
// Handle JSON POST (Elasticsearch)
@@ -369,11 +369,24 @@ async fn handle_instant_query_post(
369369
}
370370
};
371371

372-
// Parse form parameters
373-
let params: HashMap<String, String> = form_urlencoded::parse(body_str.as_bytes())
372+
debug!(
373+
"POST body prefix (first 200 chars): '{}'",
374+
&body_str.chars().take(200).collect::<String>()
375+
);
376+
377+
// Parse form parameters, falling back to treating the raw body as the
378+
// SQL query if no "query" key is found (ClickHouse native HTTP style).
379+
let mut params: HashMap<String, String> = form_urlencoded::parse(body_str.as_bytes())
374380
.into_owned()
375381
.collect();
376-
debug!("Form params extracted: {:?}", params);
382+
if !params.contains_key("query") && !body_str.is_empty() {
383+
params.clear();
384+
params.insert("query".to_string(), body_str.clone());
385+
}
386+
debug!(
387+
"POST form params keys: {:?}",
388+
params.keys().collect::<Vec<_>>()
389+
);
377390

378391
// Use adapter to parse POST request (handles form-encoded parameters)
379392
match state.adapter.parse_post_request(Form(params)).await {
@@ -385,7 +398,7 @@ async fn handle_instant_query_post(
385398
req
386399
}
387400
Err(parse_error) => {
388-
debug!("Failed to parse POST request: {:?}", parse_error);
401+
debug!("POST parse failed: {}", parse_error);
389402
return match state.adapter.format_error_response(&parse_error).await {
390403
Ok(json) => json.into_response(),
391404
Err(status) => status.into_response(),

‎asap-query-engine/src/precompute_engine/ingest_source.rs‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,10 @@ use crate::precompute_engine::worker::{extract_metric_name, parse_labels_from_se
44
use arc_swap::ArcSwap;
55
use asap_types::aggregation_config::AggregationConfig;
66
use std::collections::HashMap;
7+
use std::sync::atomic::{AtomicBool, Ordering};
78
use std::sync::Arc;
89
use std::time::Instant;
10+
use tracing::{debug, warn};
911

1012
/// Everything a source needs to push decoded samples into the worker pool.
1113
#[derive(Clone)]
@@ -89,15 +91,42 @@ pub(crate) async fn route_decoded_samples(
8991

9092
// Load agg_configs once per request (lock-free ArcSwap read).
9193
let agg_configs = ctx.agg_configs.load();
94+
95+
// On first batch: log config metrics vs sample metric to diagnose mismatches.
96+
static FIRST_BATCH_LOGGED: AtomicBool = AtomicBool::new(false);
97+
if !FIRST_BATCH_LOGGED.swap(true, Ordering::Relaxed) {
98+
if let Some(first) = samples.first() {
99+
let sample_metric = extract_metric_name(&first.labels);
100+
warn!(
101+
sample_metric,
102+
sample_labels = %first.labels,
103+
num_agg_configs = agg_configs.len(),
104+
"routing: first batch diagnostic"
105+
);
106+
for cfg in agg_configs.iter() {
107+
warn!(
108+
agg_id = cfg.aggregation_id,
109+
config_metric = %cfg.metric,
110+
config_spatial_filter = %cfg.spatial_filter,
111+
table_name = ?cfg.table_name,
112+
"routing: agg config metric"
113+
);
114+
}
115+
}
116+
}
117+
118+
let mut matched_samples: usize = 0;
92119
for s in &samples {
93120
let metric_name = extract_metric_name(&s.labels);
94121
for config in agg_configs.iter() {
95122
if config.metric != metric_name
96123
&& config.spatial_filter_normalized != metric_name
97124
&& config.spatial_filter != metric_name
125+
&& config.table_name.as_deref() != Some(metric_name)
98126
{
99127
continue;
100128
}
129+
matched_samples += 1;
101130
let group_key = extract_group_key(&s.labels, config);
102131
by_group
103132
.entry((config.aggregation_id, group_key))
@@ -106,6 +135,13 @@ pub(crate) async fn route_decoded_samples(
106135
}
107136
}
108137

138+
debug!(
139+
total_samples = samples.len(),
140+
matched_samples,
141+
groups_formed = by_group.len(),
142+
"routing: batch match summary"
143+
);
144+
109145
let messages: Vec<WorkerMessage> = by_group
110146
.into_iter()
111147
.map(

‎asap-query-engine/src/precompute_engine/json_ingest.rs‎

Lines changed: 42 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use crate::drivers::ingest::prometheus_remote_write::DecodedSample;
22
use crate::precompute_engine::ingest_source::{route_decoded_samples, IngestContext, IngestSource};
3+
use chrono::NaiveDateTime;
34
use std::time::Instant;
45
use tracing::info;
56

@@ -110,23 +111,38 @@ impl IngestSource for JsonFileIngestSource {
110111
.into()
111112
})?;
112113

113-
let raw_ts: i64 = obj
114-
.get(&config.timestamp_col)
115-
.ok_or_else(|| -> Box<dyn std::error::Error + Send + Sync> {
114+
let ts_val = obj.get(&config.timestamp_col).ok_or_else(
115+
|| -> Box<dyn std::error::Error + Send + Sync> {
116116
std::io::Error::other(format!(
117117
"timestamp column '{}' not found in JSON object",
118118
config.timestamp_col
119119
))
120120
.into()
121-
})?
122-
.as_i64()
123-
.ok_or_else(|| -> Box<dyn std::error::Error + Send + Sync> {
124-
std::io::Error::other(format!(
125-
"timestamp column '{}' is not an integer",
126-
config.timestamp_col
127-
))
128-
.into()
129-
})?;
121+
},
122+
)?;
123+
124+
// Accept integer (Unix epoch) or string datetime "YYYY-MM-DD HH:MM:SS".
125+
let raw_ts: i64 = if let Some(i) = ts_val.as_i64() {
126+
i
127+
} else if let Some(s) = ts_val.as_str() {
128+
NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S")
129+
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> {
130+
std::io::Error::other(format!(
131+
"timestamp column '{}' value {:?} is not an integer or \
132+
a parseable datetime string: {}",
133+
config.timestamp_col, s, e
134+
))
135+
.into()
136+
})?
137+
.and_utc()
138+
.timestamp()
139+
} else {
140+
return Err(std::io::Error::other(format!(
141+
"timestamp column '{}' is not an integer or string",
142+
config.timestamp_col
143+
))
144+
.into());
145+
};
130146

131147
let timestamp_ms = config.timestamp_unit.to_ms(raw_ts);
132148

@@ -140,7 +156,20 @@ impl IngestSource for JsonFileIngestSource {
140156
if i > 0 {
141157
s.push(',');
142158
}
143-
let val = obj.get(col).and_then(|v| v.as_str()).unwrap_or("");
159+
let val_owned;
160+
let val = if let Some(s) = obj.get(col).and_then(|v| v.as_str()) {
161+
s
162+
} else if let Some(v) = obj.get(col) {
163+
val_owned = v.to_string();
164+
val_owned.as_str()
165+
} else {
166+
return Err(std::io::Error::other(format!(
167+
"label column '{}' not found in JSON object (row {})",
168+
col,
169+
row_count + 1
170+
))
171+
.into());
172+
};
144173
s.push_str(col);
145174
s.push_str("=\"");
146175
s.push_str(val);

‎asap-tools/execution-utilities/benchmark/generate_queries.py‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -240,15 +240,12 @@ def generate_sql_file(
240240
start_str = (end_ts - timedelta(seconds=window_size)).strftime(
241241
"%Y-%m-%d %H:%M:%S"
242242
)
243-
label = f"T{i:03d}"
244-
245243
if window_form == "dateadd":
246244
where = f"{ts_column} BETWEEN DATEADD(s, -{window_size}, '{end_str}') AND '{end_str}'"
247245
else:
248246
where = f"{ts_column} BETWEEN '{start_str}' AND '{end_str}'"
249247

250248
lines.append(
251-
f"-- {label}: quantile window ending at {end_str}\n"
252249
f"SELECT quantile({quantile})({value_column}) FROM {table_name} "
253250
f"WHERE {where} GROUP BY {group_by_clause};"
254251
)

‎asap-tools/experiments/experiment_run_clickhouse.py‎

Lines changed: 46 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@
9393
from urllib.parse import urlparse
9494

9595
import hydra
96+
import yaml
9697
from omegaconf import DictConfig, OmegaConf
9798

9899
import constants
@@ -106,6 +107,39 @@
106107
from experiment_utils.services.misc import ControllerService
107108
from experiment_utils.services.query_engine import QueryEngineRustService
108109

110+
111+
def _inline_sql_queries_in_experiment_config(local_experiment_root_dir: str) -> None:
112+
"""Enrich the saved experiment_params.yaml by inlining SQL from sql_file references.
113+
114+
Downstream analysis scripts expect query_groups[i]["queries"] to be a list
115+
of query strings. The clickhouse config stores sql_file paths instead, so
116+
we read each file and add the queries in-place before the scripts run.
117+
"""
118+
config_path = os.path.join(
119+
local_experiment_root_dir, "experiment_config", "experiment_params.yaml"
120+
)
121+
if not os.path.exists(config_path):
122+
return
123+
with open(config_path) as f:
124+
data = yaml.safe_load(f)
125+
126+
def _expand_groups(groups):
127+
if not groups:
128+
return
129+
for group in groups:
130+
sql_file = group.get("sql_file")
131+
if sql_file and "queries" not in group:
132+
with open(sql_file) as fq:
133+
content = fq.read()
134+
group["queries"] = [s.strip() for s in content.split(";") if s.strip()]
135+
136+
_expand_groups(data.get("query_groups"))
137+
_expand_groups(data.get("sketchdb_query_groups"))
138+
139+
with open(config_path, "w") as f:
140+
yaml.dump(data, f, allow_unicode=True)
141+
142+
109143
# Register resolvers used by config.yaml interpolation.
110144
OmegaConf.register_new_resolver(
111145
"local_experiment_dir", lambda: constants.LOCAL_EXPERIMENT_DIR
@@ -234,6 +268,7 @@ def main(cfg: DictConfig) -> None:
234268
)
235269

236270
sync.copy_experiment_config(cfg.experiment_params, local_experiment_root_dir)
271+
_inline_sql_queries_in_experiment_config(local_experiment_root_dir)
237272

238273
# --- dataset config ---
239274
ep = cfg.experiment_params
@@ -339,6 +374,11 @@ def main(cfg: DictConfig) -> None:
339374

340375
if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME:
341376
# --- sketchdb mode: precompute engine + JSON ingest + ClickHouse fallback ---
377+
# Kill any leftover query_engine_rust from a previous run (mirrors
378+
# query_engine_service.stop() at the top of the e2e mode loop).
379+
QueryEngineRustService(
380+
provider=provider, use_container=False, node_offset=node_offset
381+
).stop()
342382
# Mirrors experiment_run_e2e.py: planner runs first and generates
343383
# streaming_config.yaml + inference_config.yaml into controller_output_dir,
344384
# then the query engine starts reading from that same directory.
@@ -438,7 +478,9 @@ def main(cfg: DictConfig) -> None:
438478
provider=provider,
439479
node_offset=node_offset,
440480
config_file=controller_client_config,
441-
output_dir=experiment_output_dir,
481+
output_dir=os.path.join(
482+
experiment_output_dir, "prometheus_client_output"
483+
),
442484
use_container=use_container,
443485
parallel=parallel,
444486
)
@@ -465,7 +507,9 @@ def main(cfg: DictConfig) -> None:
465507
provider=provider,
466508
node_offset=node_offset,
467509
config_file=controller_client_config,
468-
output_dir=experiment_output_dir,
510+
output_dir=os.path.join(
511+
experiment_output_dir, "prometheus_client_output"
512+
),
469513
use_container=use_container,
470514
parallel=parallel,
471515
)

0 commit comments

Comments
 (0)