fix(planner, tools): use milliseconds for repetition_delay (#449)

* fix(planner, tools): use milliseconds for repetition_delay

* fix(planner): change more variables to ms precision

* more fixes

* fixes to python infra scripts

* fixed more bugs

* removed stale codel

* small fixes from CR

* removed unnecessary deny_unknown_fields

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: milindsrivastava1997

asap-planner-rs/docker-compose.yml.j2

@@ -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

asap-planner-rs/src/config/input.rs

@@ -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)]
@@ -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 {
@@ -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,
 }
 
@@ -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,

asap-planner-rs/src/elastic_dsl/generator.rs

@@ -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(
@@ -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
             )));
         }
     }
@@ -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(),

asap-planner-rs/src/main.rs

@@ -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
@@ -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(
@@ -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)?;
         }

asap-planner-rs/src/optimizer/pipeline.rs

@@ -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

asap-planner-rs/src/planner/cleanup.rs

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

asap-planner-rs/src/planner/elastic_dsl.rs

@@ -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,
@@ -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,
@@ -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,
@@ -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,
         };
@@ -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)?,
             )
         };

asap-planner-rs/src/planner/sql.rs

@@ -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,
@@ -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,
@@ -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()));
@@ -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)?;
@@ -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)?,
             )
         };
@@ -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,

asap-planner-rs/src/promql/controller.rs

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