Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion asap-planner-rs/src/planner/elastic_dsl.rs
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,12 @@ impl ElasticSingleQueryProcessor {
let (treatment_type, statistics) = get_elastic_statistics(&query_info.aggregation)?;

let t_repeat_ms = self.t_repeat_ms;
if t_repeat_ms < self.data_ingestion_interval_ms {
return Err(ControllerError::UnsupportedElasticDSLQuery(format!(
"repetition interval {}ms is shorter than the data ingestion interval {}ms",
t_repeat_ms, self.data_ingestion_interval_ms
)));
}

// Validate and resolve the time-range predicate before building configs.
let time_field = self.index_schema.time_field.clone();
Expand All @@ -83,7 +89,12 @@ impl ElasticSingleQueryProcessor {
))
}
Some(tr) => {
let duration = tr.duration_ms().unwrap_or(t_repeat_ms);
let duration = tr.duration_ms().ok_or_else(|| {
ControllerError::UnsupportedElasticDSLQuery(
"time-range predicate must have finite, valid gte and lte bounds"
.to_string(),
)
})?;
if duration < self.data_ingestion_interval_ms {
return Err(ControllerError::UnsupportedElasticDSLQuery(format!(
"time-range duration {}ms is shorter than the data ingestion interval {}ms",
Expand All @@ -94,6 +105,12 @@ impl ElasticSingleQueryProcessor {
let window = if duration == self.data_ingestion_interval_ms {
self.data_ingestion_interval_ms
} else {
if duration < t_repeat_ms {
return Err(ControllerError::UnsupportedElasticDSLQuery(format!(
"time-range duration {}ms is shorter than the repetition interval {}ms",
duration, t_repeat_ms
)));
}
t_repeat_ms
};
(duration, window)
Expand Down
74 changes: 72 additions & 2 deletions asap-planner-rs/tests/elastic_dsl_integration.rs
Original file line number Diff line number Diff line change
Expand Up @@ -524,6 +524,15 @@ fn time_range_query(duration: &str) -> String {
)
}

fn one_sided_time_range_query(duration: &str) -> String {
format!(
r#"{{
"aggs": {{ "sum_cpu": {{ "sum": {{ "field": "cpu_usage" }} }} }},
"query": {{ "bool": {{ "filter": [{{ "range": {{ "@timestamp": {{ "gte": "now-{duration}" }} }} }}] }} }}
}}"#
)
}

const QUERY_NO_TIME_RANGE: &str = r#"{
"aggs": { "sum_cpu": { "sum": { "field": "cpu_usage" } } }
}"#;
Expand Down Expand Up @@ -559,6 +568,21 @@ fn time_range_shorter_than_ingestion_interval_is_rejected() {
));
}

#[test]
fn t_repeat_shorter_than_ingestion_interval_is_rejected() {
let result = try_elastic_with_interval(
"metrics",
"\"@timestamp\"",
&time_range_query("1m"),
0,
15_000,
);
assert!(matches!(
result,
Err(ControllerError::UnsupportedElasticDSLQuery(_))
));
}

#[test]
fn time_range_equal_to_ingestion_interval_uses_interval_as_window_size() {
// data_ingestion_interval_ms = 15_000ms (15s), query range = 15s → Spatial: window = interval
Expand All @@ -574,8 +598,40 @@ fn time_range_equal_to_ingestion_interval_uses_interval_as_window_size() {
}

#[test]
fn time_range_longer_than_ingestion_interval_uses_t_repeat_as_window_size() {
// data_ingestion_interval_ms = 15_000ms (15s), query range = 5m > 15s → Temporal: window = t_repeat_ms
fn time_range_shorter_than_t_repeat_is_rejected() {
// data_ingestion_interval_ms = 15_000ms (15s), query range = 1m, t_repeat = 5m.
// A temporal precompute window must not outlive the query range it serves.
let t_repeat_ms = 300_000;
let result = try_elastic_with_interval(
"metrics",
"\"@timestamp\"",
&time_range_query("1m"),
t_repeat_ms,
15_000,
);
assert!(matches!(
result,
Err(ControllerError::UnsupportedElasticDSLQuery(_))
));
}

#[test]
fn one_sided_time_range_is_rejected() {
let result = try_elastic_with_interval(
"metrics",
"\"@timestamp\"",
&one_sided_time_range_query("1m"),
300_000,
15_000,
);
assert!(matches!(
result,
Err(ControllerError::UnsupportedElasticDSLQuery(_))
));
}

#[test]
fn time_range_equal_to_t_repeat_uses_t_repeat_as_window_size() {
let t_repeat_ms = 300_000;
let out = try_elastic_with_interval(
"metrics",
Expand All @@ -587,3 +643,17 @@ fn time_range_longer_than_ingestion_interval_uses_t_repeat_as_window_size() {
.unwrap();
assert!(out.all_tumbling_window_sizes_eq(t_repeat_ms));
}

#[test]
fn time_range_at_least_t_repeat_uses_t_repeat_as_window_size() {
let t_repeat_ms = 300_000;
let out = try_elastic_with_interval(
"metrics",
"\"@timestamp\"",
&time_range_query("10m"),
t_repeat_ms,
15_000,
)
.unwrap();
assert!(out.all_tumbling_window_sizes_eq(t_repeat_ms));
}
Loading