refactor: unify prometheus_scrape_interval and data_ingestion_interval

Author: milindsrivastava1997

asap-common/dependencies/py/promql_utilities/promql_utilities/query_logics/parsing.py

@@ -35,7 +35,6 @@ def get_statistics_to_compute(
         # template_config.tumblingWindowSize = self.t_repeat
     elif query_pattern_type == QueryPatternType.ONLY_SPATIAL:
         statistic_to_compute = query_pattern_match.tokens["aggregation"]["op"]
-        # template_config.tumblingWindowSize = self.prometheus_scrape_interval
     else:
         raise ValueError("Invalid query pattern type")
 
@@ -55,7 +54,7 @@ def get_spatial_aggregation_output_labels(
     # Fixing issue https://github.com/ProjectASAP/asap-internal/issues/24
     if aggregation_modifier is None:
         return KeyByLabelNames([])
-    
+
     if aggregation_modifier.type == aggregation_modifier.type.By:
         aggregation_modifier_labels = KeyByLabelNames(aggregation_modifier.labels)
     elif aggregation_modifier.type == aggregation_modifier.type.Without:

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

@@ -8,12 +8,11 @@ services:
     command: [
       "--input_config", "/app/input/config.yaml",
       "--output_dir", "/app/output",
-      "--prometheus_scrape_interval_ms", "{{ prometheus_scrape_interval_ms }}",
+      "--data-ingestion-interval-ms", "{{ data_ingestion_interval_ms }}",
       "--streaming_engine", "{{ streaming_engine }}",
       "--prometheus-url", "{{ prometheus_url }}",
       "--query-language", "{{ query_language }}"{% if punting %},
-      "--enable-punting"{% endif %}{% if data_ingestion_interval_ms is not none %},
-      "--data-ingestion-interval-ms", "{{ data_ingestion_interval_ms }}"{% endif %}
+      "--enable-punting"{% endif %}
     ]
     network_mode: "host"
     restart: no

asap-planner-rs/src/bin/optimizer_cli.rs

@@ -20,8 +20,8 @@ struct Args {
     #[arg(long = "input_config")]
     input_config: PathBuf,
 
-    #[arg(long = "prometheus_scrape_interval_ms")]
-    prometheus_scrape_interval_ms: u64,
+    #[arg(long = "data-ingestion-interval-ms")]
+    data_ingestion_interval_ms: u64,
 
     /// Placeholder arrival rate (items/sec) applied uniformly to every candidate's
     /// IngestCost. Real per-config rates aren't wired up yet — see the open TODOs
@@ -56,12 +56,8 @@ fn main() -> anyhow::Result<()> {
     let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
     let schema = config.schema_from_hints();
 
-    let (streaming, inference) = run_greedy_pipeline(
-        &config,
-        &schema,
-        args.prometheus_scrape_interval_ms,
-        args.rho,
-    );
+    let (streaming, inference) =
+        run_greedy_pipeline(&config, &schema, args.data_ingestion_interval_ms, args.rho);
 
     let deployed = streaming.get_all_aggregation_configs();
     println!("=== Deployed streaming configs: {} ===", deployed.len());

asap-planner-rs/src/lib.rs

@@ -35,7 +35,7 @@ pub enum StreamingEngine {
 
 #[derive(Debug, Clone)]
 pub struct RuntimeOptions {
-    pub prometheus_scrape_interval_ms: u64,
+    pub data_ingestion_interval_ms: u64,
     pub streaming_engine: StreamingEngine,
     pub enable_punting: bool,
     pub range_duration_ms: u64,

asap-planner-rs/src/main.rs

@@ -20,9 +20,6 @@ struct Args {
     #[arg(long = "output_dir")]
     output_dir: PathBuf,
 
-    #[arg(long = "prometheus_scrape_interval_ms", required = false)]
-    prometheus_scrape_interval_ms: Option<u64>,
-
     /// Base URL of the Prometheus instance used to auto-infer metric label sets.
     /// Optional: when provided, the planner queries Prometheus for label discovery.
     /// When absent, labels are taken from the `metrics` hint in the config file.
@@ -86,11 +83,11 @@ fn main() -> anyhow::Result<()> {
 
     match args.query_language {
         QueryLanguage::promql => {
-            let scrape_interval_ms = args.prometheus_scrape_interval_ms.ok_or_else(|| {
-                anyhow::anyhow!("--prometheus_scrape_interval_ms is required for PromQL mode")
+            let scrape_interval_ms = args.data_ingestion_interval_ms.ok_or_else(|| {
+                anyhow::anyhow!("--data-ingestion-interval-ms is required for PromQL mode")
             })?;
             let opts = RuntimeOptions {
-                prometheus_scrape_interval_ms: scrape_interval_ms,
+                data_ingestion_interval_ms: scrape_interval_ms,
                 streaming_engine: engine,
                 enable_punting: args.enable_punting,
                 range_duration_ms: args.range_duration_ms,

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

@@ -72,7 +72,7 @@ pub(crate) fn parse_binary_arms(query: &str) -> Option<(BinaryArm, BinaryArm)> {
 pub struct SingleQueryProcessor {
     query: String,
     t_repeat_ms: u64,
-    prometheus_scrape_interval_ms: u64,
+    data_ingestion_interval_ms: u64,
     metric_schema: PromQLSchema,
     #[allow(dead_code)]
     streaming_engine: StreamingEngine,
@@ -87,7 +87,7 @@ impl SingleQueryProcessor {
     pub fn new(
         query: String,
         t_repeat_ms: u64,
-        prometheus_scrape_interval_ms: u64,
+        data_ingestion_interval_ms: u64,
         metric_schema: PromQLSchema,
         streaming_engine: StreamingEngine,
         sketch_parameters: Option<SketchParameterOverrides>,
@@ -98,7 +98,7 @@ impl SingleQueryProcessor {
         Self {
             query,
             t_repeat_ms,
-            prometheus_scrape_interval_ms,
+            data_ingestion_interval_ms,
             metric_schema,
             streaming_engine,
             sketch_parameters,
@@ -159,7 +159,7 @@ impl SingleQueryProcessor {
         SingleQueryProcessor::new(
             arm_query,
             self.t_repeat_ms,
-            self.prometheus_scrape_interval_ms,
+            self.data_ingestion_interval_ms,
             self.metric_schema.clone(),
             self.streaming_engine,
             self.sketch_parameters.clone(),
@@ -199,7 +199,7 @@ impl SingleQueryProcessor {
                     | PromQLFunction::QuantileOverTime)
             ) {
                 let num_data_points =
-                    self.t_repeat_ms as f64 / self.prometheus_scrape_interval_ms as f64;
+                    self.t_repeat_ms as f64 / self.data_ingestion_interval_ms as f64;
                 if num_data_points < 60.0 {
                     return false;
                 }
@@ -248,7 +248,7 @@ impl SingleQueryProcessor {
         set_window_parameters(
             pattern_type,
             self.t_repeat_ms,
-            self.prometheus_scrape_interval_ms,
+            self.data_ingestion_interval_ms,
             "any", // aggregation_type doesn't matter (sliding always false)
             self.step_ms,
             &mut window_cfg,

asap-planner-rs/src/planner/window.rs

@@ -20,7 +20,7 @@ pub fn should_use_sliding_window(
 pub fn set_window_parameters(
     query_pattern_type: QueryPatternType,
     t_repeat_ms: u64,
-    prometheus_scrape_interval_ms: u64,
+    data_ingestion_interval_ms: u64,
     aggregation_type: &str,
     step_ms: u64,
     config: &mut IntermediateWindowConfig,
@@ -31,15 +31,15 @@ pub fn set_window_parameters(
     set_tumbling_window_parameters(
         query_pattern_type,
         effective_repeat,
-        prometheus_scrape_interval_ms,
+        data_ingestion_interval_ms,
         config,
     );
 }
 
 fn set_tumbling_window_parameters(
     query_pattern_type: QueryPatternType,
     effective_repeat: u64,
-    prometheus_scrape_interval_ms: u64,
+    data_ingestion_interval_ms: u64,
     config: &mut IntermediateWindowConfig,
 ) {
     match query_pattern_type {
@@ -49,8 +49,8 @@ fn set_tumbling_window_parameters(
             config.window_type = WindowType::Tumbling;
         }
         QueryPatternType::OnlySpatial => {
-            config.window_size_ms = prometheus_scrape_interval_ms;
-            config.slide_interval_ms = prometheus_scrape_interval_ms;
+            config.window_size_ms = data_ingestion_interval_ms;
+            config.slide_interval_ms = data_ingestion_interval_ms;
             config.window_type = WindowType::Tumbling;
         }
     }

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

@@ -114,7 +114,7 @@ impl Controller {
     ) -> Result<Self, ControllerError> {
         let entries = query_log::parse_log_file(log_path)?;
         let (instants, ranges) =
-            query_log::infer_queries(&entries, opts.prometheus_scrape_interval_ms);
+            query_log::infer_queries(&entries, opts.data_ingestion_interval_ms);
         let config = query_log::to_controller_config(instants, ranges);
         let all_queries: Vec<String> = config
             .query_groups
@@ -139,7 +139,7 @@ impl Controller {
     ) -> Result<Self, ControllerError> {
         let entries = query_log::parse_log_file(log_path)?;
         let (instants, ranges) =
-            query_log::infer_queries(&entries, opts.prometheus_scrape_interval_ms);
+            query_log::infer_queries(&entries, opts.data_ingestion_interval_ms);
         let config = query_log::to_controller_config(instants, ranges);
         Ok(Self {
             config,

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

@@ -60,7 +60,7 @@ pub fn generate_plan(
             let processor = SingleQueryProcessor::new(
                 query_string.clone(),
                 qg.repetition_delay_ms,
-                opts.prometheus_scrape_interval_ms,
+                opts.data_ingestion_interval_ms,
                 metric_schema.clone(),
                 opts.streaming_engine,
                 controller_config.sketch_parameters.clone(),

asap-planner-rs/tests/integration.rs

@@ -6,7 +6,7 @@ use std::path::Path;
 
 fn arroyo_opts() -> RuntimeOptions {
     RuntimeOptions {
-        prometheus_scrape_interval_ms: 15_000,
+        data_ingestion_interval_ms: 15_000,
         streaming_engine: StreamingEngine::Arroyo,
         enable_punting: false,
         range_duration_ms: 0,
@@ -598,7 +598,7 @@ fn sub_second_scrape_interval_window_size_equals_scrape_interval() {
     // OnlySpatial query: window size = scrape interval (planner/window.rs).
     // 100ms would be indistinguishable from 0 under the old seconds-only model.
     let opts = RuntimeOptions {
-        prometheus_scrape_interval_ms: 100,
+        data_ingestion_interval_ms: 100,
         ..arroyo_opts()
     };
     let c = Controller::from_file_with_schema(
@@ -617,7 +617,7 @@ fn sub_second_scrape_interval_round_trips_through_generated_yaml() {
     // under the renamed wire key (windowSizeMs, not windowSize) — not silently
     // rounded or truncated.
     let opts = RuntimeOptions {
-        prometheus_scrape_interval_ms: 100,
+        data_ingestion_interval_ms: 100,
         ..arroyo_opts()
     };
     let c = Controller::from_file_with_schema(