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
4 changes: 2 additions & 2 deletions asap-planner-rs/docker-compose.yml.j2
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,8 @@ services:
"--streaming_engine", "{{ streaming_engine }}",
"--prometheus-url", "{{ prometheus_url }}",
"--query-language", "{{ query_language }}"{% if punting %},
"--enable-punting"{% endif %}{% if data_ingestion_interval is not none %},
"--data-ingestion-interval", "{{ data_ingestion_interval }}"{% endif %}
"--enable-punting"{% endif %}{% if data_ingestion_interval_ms is not none %},
"--data-ingestion-interval-ms", "{{ data_ingestion_interval_ms }}"{% endif %}
]
network_mode: "host"
restart: no
21 changes: 19 additions & 2 deletions asap-planner-rs/src/config/input.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ use asap_types::streaming_config::StreamingConfig;
use asap_types::PromQLSchema;
use promql_utilities::data_model::KeyByLabelNames;
use serde::Deserialize;
use tracing::warn;

#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
Expand All @@ -30,6 +31,22 @@ pub struct ControllerConfig {
}

impl ControllerConfig {
/// Warn if any query group has both SLAs at 0.0 (the serde Default),
/// which indicates `controller_options` was omitted from the config.
pub fn warn_default_slas(&self) {
for qg in &self.query_groups {
let opts = &qg.controller_options;
if opts.accuracy_sla == 0.0 && opts.latency_sla == 0.0 {
warn!(
query_group_id = ?qg.id,
"controller_options not set in query group; \
accuracy_sla=0.0 and latency_sla=0.0 will be used — \
add controller_options to your config"
);
}
}
}

/// Build a `PromQLSchema` from the `metrics` hints in this config.
/// Returns an empty schema if no hints are present.
pub fn schema_from_hints(&self) -> PromQLSchema {
Expand Down Expand Up @@ -133,7 +150,7 @@ pub struct SQLControllerConfig {
pub struct SQLQueryGroup {
pub id: Option<u32>,
pub queries: Vec<String>,
pub repetition_delay: u64,
pub repetition_delay_ms: u64,
pub controller_options: ControllerOptions,
}

Expand All @@ -157,7 +174,7 @@ pub struct ElasticDSLControllerConfig {
pub struct ElasticDSLQueryGroup {
pub id: Option<u32>,
pub queries: Vec<String>,
pub repetition_delay: u64,
pub repetition_delay_ms: u64,
pub index: String,
pub time_field: String,
pub controller_options: ControllerOptions,
Expand Down
14 changes: 7 additions & 7 deletions asap-planner-rs/src/elastic_dsl/generator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ impl ElasticIndexSchemaBuilder {

pub struct ElasticRuntimeOptions {
pub streaming_engine: StreamingEngine,
pub data_ingestion_interval: u64,
pub data_ingestion_interval_ms: u64,
}

pub fn generate_elastic_plan(
Expand All @@ -60,12 +60,12 @@ pub fn generate_elastic_plan(
.and_then(|c| c.policy)
.unwrap_or(CleanupPolicy::ReadBased);

// Validate T % data_ingestion_interval == 0
// Validate T % data_ingestion_interval_ms == 0
for qg in &config.query_groups {
if qg.repetition_delay % opts.data_ingestion_interval != 0 {
if qg.repetition_delay_ms % opts.data_ingestion_interval_ms != 0 {
return Err(ControllerError::PlannerError(format!(
"repetition_delay {} is not a multiple of data_ingestion_interval {}",
qg.repetition_delay, opts.data_ingestion_interval
"repetition_delay_ms {} is not a multiple of data_ingestion_interval_ms {}",
qg.repetition_delay_ms, opts.data_ingestion_interval_ms
)));
}
}
Expand Down Expand Up @@ -111,8 +111,8 @@ pub fn generate_elastic_plan(
for query_string in &qg.queries {
let processor = ElasticSingleQueryProcessor::new(
query_string.clone(),
qg.repetition_delay,
opts.data_ingestion_interval,
qg.repetition_delay_ms,
opts.data_ingestion_interval_ms,
index_schema_builders[&qg.index].clone(),
opts.streaming_engine,
config.sketch_parameters.clone(),
Expand Down
18 changes: 10 additions & 8 deletions asap-planner-rs/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,8 @@ struct Args {
#[arg(long = "query-language", value_enum, default_value = "promql")]
query_language: QueryLanguage,

#[arg(long = "data-ingestion-interval", required = false)]
data_ingestion_interval: Option<u64>,
#[arg(long = "data-ingestion-interval-ms", required = false)]
data_ingestion_interval_ms: Option<u64>,

/// ClickHouse base URL for auto-inferring metadata_columns when not listed
/// in the config file. Example: http://localhost:8123
Expand Down Expand Up @@ -123,16 +123,16 @@ fn main() -> anyhow::Result<()> {
controller.generate_to_dir(&args.output_dir)?;
}
QueryLanguage::sql | QueryLanguage::elastic_sql => {
let interval = args.data_ingestion_interval.ok_or_else(|| {
anyhow::anyhow!("--data-ingestion-interval is required for SQL mode")
let interval = args.data_ingestion_interval_ms.ok_or_else(|| {
anyhow::anyhow!("--data-ingestion-interval-ms is required for SQL mode")
})?;
let config_path = args
.input_config
.ok_or_else(|| anyhow::anyhow!("--input_config is required for SQL mode"))?;
let opts = SQLRuntimeOptions {
streaming_engine: engine,
query_evaluation_time: None,
data_ingestion_interval: interval,
data_ingestion_interval_ms: interval,
};
let controller = match args.clickhouse_url {
Some(ref url) => SQLController::from_file_with_discovery(
Expand All @@ -146,15 +146,17 @@ fn main() -> anyhow::Result<()> {
controller.generate_to_dir(&args.output_dir)?;
}
QueryLanguage::elastic_querydsl => {
let interval = args.data_ingestion_interval.ok_or_else(|| {
anyhow::anyhow!("--data-ingestion-interval is required for Elasticsearch DSL mode")
let interval = args.data_ingestion_interval_ms.ok_or_else(|| {
anyhow::anyhow!(
"--data-ingestion-interval-ms is required for Elasticsearch DSL mode"
)
})?;
let config_path = args.input_config.ok_or_else(|| {
anyhow::anyhow!("--input_config is required for Elasticsearch DSL mode")
})?;
let opts = ElasticRuntimeOptions {
streaming_engine: engine,
data_ingestion_interval: interval,
data_ingestion_interval_ms: interval,
};
ElasticController::from_file(&config_path, opts)?.generate_to_dir(&args.output_dir)?;
}
Expand Down
2 changes: 1 addition & 1 deletion asap-planner-rs/src/optimizer/pipeline.rs
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ pub fn run_greedy_pipeline(
}

/// Convert a `ControllerConfig`'s query groups into a flat list of RQEs.
/// Each (query, repetition_delay) pair becomes one RQE.
/// Each (query, repetition_delay_ms) pair becomes one RQE.
fn config_to_rqes(config: &ControllerConfig) -> Vec<RQE> {
config
.query_groups
Expand Down
12 changes: 6 additions & 6 deletions asap-planner-rs/src/planner/cleanup.rs
Original file line number Diff line number Diff line change
Expand Up @@ -75,19 +75,19 @@ pub fn get_cleanup_param(
/// SQL cleanup param — SQL queries are always instant (no range_duration/step).
pub fn get_sql_cleanup_param(
cleanup_policy: CleanupPolicy,
t_lookback: u64,
t_repeat: u64,
t_lookback_ms: u64,
t_repeat_ms: u64,
) -> Result<u64, String> {
match cleanup_policy {
CleanupPolicy::CircularBuffer | CleanupPolicy::ReadBased => {
if t_repeat == 0 {
if t_repeat_ms == 0 {
return Err(
"repetition_delay must be > 0 for cleanup param calculation; \
set a non-zero repetition_delay in your query group config"
"repetition_delay_ms must be > 0 for cleanup param calculation; \
set a non-zero repetition_delay_ms in your query group config"
.to_string(),
);
}
Ok(t_lookback.div_ceil(t_repeat))
Ok(t_lookback_ms.div_ceil(t_repeat_ms))
}
CleanupPolicy::NoCleanup => {
Err("NoCleanup policy should not call get_sql_cleanup_param".to_string())
Expand Down
22 changes: 9 additions & 13 deletions asap-planner-rs/src/planner/elastic_dsl.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@ use indexmap::IndexSet;

pub struct ElasticSingleQueryProcessor {
query_string: String,
t_repeat: u64,
t_repeat_ms: u64,
#[allow(dead_code)]
data_ingestion_interval: u64,
data_ingestion_interval_ms: u64,
index_schema: ElasticIndexSchemaBuilder,
#[allow(dead_code)]
streaming_engine: StreamingEngine,
Expand All @@ -33,17 +33,17 @@ impl ElasticSingleQueryProcessor {
#[allow(clippy::too_many_arguments)]
pub fn new(
query_string: String,
t_repeat: u64,
data_ingestion_interval: u64,
t_repeat_ms: u64,
data_ingestion_interval_ms: u64,
index_schema: ElasticIndexSchemaBuilder,
streaming_engine: StreamingEngine,
sketch_parameters: Option<SketchParameterOverrides>,
cleanup_policy: CleanupPolicy,
) -> Self {
Self {
query_string,
t_repeat,
data_ingestion_interval,
t_repeat_ms,
data_ingestion_interval_ms,
index_schema,
streaming_engine,
sketch_parameters,
Expand All @@ -65,11 +65,7 @@ impl ElasticSingleQueryProcessor {
// Get aggregation type and statistics
let (treatment_type, statistics) = get_elastic_statistics(&query_info.aggregation)?;

// Build window config (always tumbling for Elasticsearch queries).
// self.t_repeat is seconds (Elastic DSL is out of scope for the ms-precision
// rename, see issue #427) — IntermediateWindowConfig is now ms-typed throughout,
// so convert at this boundary, same as planner::sql::compute_sql_window.
let t_repeat_ms = self.t_repeat * 1000;
let t_repeat_ms = self.t_repeat_ms;
let window_cfg = IntermediateWindowConfig {
window_size_ms: t_repeat_ms,
slide_interval_ms: t_repeat_ms,
Expand Down Expand Up @@ -134,7 +130,7 @@ impl ElasticSingleQueryProcessor {
_ => false,
})
.and_then(|p| range_query_to_time_range(p, 0));
let t_lookback = match time_range {
let t_lookback_ms = match time_range {
Some(tr) => tr.duration_ms().unwrap_or(t_repeat_ms),
None => t_repeat_ms,
};
Expand All @@ -144,7 +140,7 @@ impl ElasticSingleQueryProcessor {
None
} else {
Some(
get_sql_cleanup_param(self.cleanup_policy, t_lookback, t_repeat_ms)
get_sql_cleanup_param(self.cleanup_policy, t_lookback_ms, t_repeat_ms)
.map_err(ControllerError::PlannerError)?,
)
};
Expand Down
48 changes: 26 additions & 22 deletions asap-planner-rs/src/planner/sql.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@ use crate::StreamingEngine;

pub struct SQLSingleQueryProcessor {
query_string: String,
t_repeat: u64,
data_ingestion_interval: u64,
t_repeat_ms: u64,
data_ingestion_interval_ms: u64,
table_definitions: Vec<TableDefinition>,
#[allow(dead_code)]
streaming_engine: StreamingEngine,
Expand All @@ -33,17 +33,17 @@ impl SQLSingleQueryProcessor {
#[allow(clippy::too_many_arguments)]
pub fn new(
query_string: String,
t_repeat: u64,
data_ingestion_interval: u64,
t_repeat_ms: u64,
data_ingestion_interval_ms: u64,
table_definitions: Vec<TableDefinition>,
streaming_engine: StreamingEngine,
sketch_parameters: Option<SketchParameterOverrides>,
cleanup_policy: CleanupPolicy,
) -> Self {
Self {
query_string,
t_repeat,
data_ingestion_interval,
t_repeat_ms,
data_ingestion_interval_ms,
table_definitions,
streaming_engine,
sketch_parameters,
Expand Down Expand Up @@ -72,8 +72,12 @@ impl SQLSingleQueryProcessor {
})?;

// Match query to pattern
let sql_query = SQLPatternMatcher::new(schema.clone(), self.data_ingestion_interval as f64)
.query_info_to_pattern(&qdata);
// SQLPatternMatcher.scrape_interval is in seconds (SQL timestamps are seconds-based).
let sql_query = SQLPatternMatcher::new(
schema.clone(),
self.data_ingestion_interval_ms as f64 / 1000.0,
)
.query_info_to_pattern(&qdata);

if !sql_query.is_valid() {
return Err(ControllerError::SqlParse(sql_query.msg.unwrap_or_default()));
Expand All @@ -97,8 +101,11 @@ impl SQLSingleQueryProcessor {
let value_column = agg_info.get_value_column_name().to_string();

// Compute window
let window_cfg =
compute_sql_window(query_type, self.data_ingestion_interval, self.t_repeat);
let window_cfg = compute_sql_window(
query_type,
self.data_ingestion_interval_ms,
self.t_repeat_ms,
);

// Get all metadata columns for the table
let all_metadata = get_all_metadata_columns(&self.table_definitions, table_name)?;
Expand Down Expand Up @@ -157,16 +164,17 @@ impl SQLSingleQueryProcessor {
}
}

let t_lookback = match query_type {
QueryType::Spatial => self.data_ingestion_interval,
_ => sql_query.query_data[0].time_info.get_duration() as u64,
let t_lookback_ms = match query_type {
QueryType::Spatial => self.data_ingestion_interval_ms,
// SQLPatternParser always produces second-based durations; convert to ms.
_ => (sql_query.query_data[0].time_info.get_duration() * 1000.0).round() as u64,
};

let cleanup_param = if self.cleanup_policy == CleanupPolicy::NoCleanup {
None
} else {
Some(
get_sql_cleanup_param(self.cleanup_policy, t_lookback, self.t_repeat)
get_sql_cleanup_param(self.cleanup_policy, t_lookback_ms, self.t_repeat_ms)
.map_err(ControllerError::PlannerError)?,
)
};
Expand Down Expand Up @@ -215,16 +223,12 @@ fn get_sql_statistics(name: &str) -> Result<Vec<Statistic>, ControllerError> {

fn compute_sql_window(
query_type: &QueryType,
data_ingestion_interval: u64,
t_repeat: u64,
data_ingestion_interval_ms: u64,
t_repeat_ms: u64,
) -> IntermediateWindowConfig {
// data_ingestion_interval and t_repeat are seconds (SQL mode is out of scope for the
// ms-precision rename, see issue #427) — IntermediateWindowConfig is now ms-typed
// throughout, so convert at this boundary, same as the pre-existing pattern in
// asap-query-engine/src/engines/simple_engine/sql.rs.
let window_size_ms = match query_type {
QueryType::Spatial => data_ingestion_interval * 1000,
_ => t_repeat * 1000,
QueryType::Spatial => data_ingestion_interval_ms,
_ => t_repeat_ms,
};
IntermediateWindowConfig {
window_size_ms,
Expand Down
3 changes: 3 additions & 0 deletions asap-planner-rs/src/promql/controller.rs
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ impl Controller {
) -> Result<Self, ControllerError> {
let yaml_str = std::fs::read_to_string(path)?;
let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
config.warn_default_slas();
let all_queries: Vec<String> = config
.query_groups
.iter()
Expand Down Expand Up @@ -78,6 +79,7 @@ impl Controller {
) -> Result<Self, ControllerError> {
let yaml_str = std::fs::read_to_string(path)?;
let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
config.warn_default_slas();
Ok(Self {
config,
schema,
Expand All @@ -92,6 +94,7 @@ impl Controller {
opts: RuntimeOptions,
) -> Result<Self, ControllerError> {
let config: ControllerConfig = serde_yaml::from_str(yaml)?;
config.warn_default_slas();
Ok(Self {
config,
schema,
Expand Down
Loading
Loading