refactor: unify prometheus_scrape_interval and data_ingestion_interval

.github/workflows/python.yml

@@ -203,6 +203,6 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         pip install -e asap-common/dependencies/py/promql_utilities/
-        pip install pytest pyyaml jinja2
+        pip install pytest pyyaml jinja2 loguru requests
     - name: Run pytest
       run: python -m pytest asap-summary-ingest/tests/ -v

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(

asap-query-engine/examples/engine_config.yaml

@@ -12,8 +12,8 @@
 output_dir: "./output"
 log_level: "INFO"                # DEBUG | INFO | WARN | ERROR (also respects RUST_LOG)
 
-# Prometheus scrape interval in milliseconds. Used by the query tracker and planner.
-prometheus_scrape_interval_ms: 15000
+# Data ingestion interval in milliseconds. Used by the query tracker and planner.
+data_ingestion_interval_ms: 15000
 
 # Processing backend. Must be consistent with the ingest source chosen below:
 #   arroyo     → requires ingest.type=kafka

asap-query-engine/src/engine_config.rs

@@ -31,8 +31,8 @@ pub fn check_config(config: &EngineConfig) -> Result<(), String> {
         return Err("ingest.ts_step_ms is required when ingest.timestamp_col is not set".into());
     }
 
-    if config.prometheus_scrape_interval_ms == 0 {
-        return Err("prometheus_scrape_interval_ms must be greater than 0".into());
+    if config.data_ingestion_interval_ms == 0 {
+        return Err("data_ingestion_interval_ms must be greater than 0".into());
     }
 
     if config.query_tracker.enabled && !matches!(config.backend, BackendConfig::Prometheus { .. }) {
@@ -47,7 +47,7 @@ pub fn check_config(config: &EngineConfig) -> Result<(), String> {
 pub struct EngineConfig {
     pub output_dir: String,
     pub log_level: String,
-    pub prometheus_scrape_interval_ms: u64,
+    pub data_ingestion_interval_ms: u64,
     pub streaming_engine: StreamingEngine,
     pub http_server: HttpServerSettings,
     pub backend: BackendConfig,
@@ -65,7 +65,7 @@ impl Default for EngineConfig {
         Self {
             output_dir: "./output".to_string(),
             log_level: "INFO".to_string(),
-            prometheus_scrape_interval_ms: 15_000,
+            data_ingestion_interval_ms: 15_000,
             streaming_engine: StreamingEngine::Precompute,
             http_server: HttpServerSettings::default(),
             backend: BackendConfig::default(),
@@ -522,7 +522,7 @@ streaming_engine: "precompute"
 ingest:
   type: "http_remote_write"
   port: 9090
-prometheus_scrape_interval_ms: 0
+data_ingestion_interval_ms: 0
 output_dir: "./output"
 "#;
         let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();

asap-query-engine/src/engines/simple_engine/mod.rs

@@ -136,7 +136,7 @@ pub struct SimpleEngine {
     /// Updated at runtime via update_streaming_config(). Readers briefly lock to
     /// clone the Arc pointer, then use without holding the lock.
     streaming_config: RwLock<Arc<StreamingConfig>>,
-    prometheus_scrape_interval_ms: u64,
+    data_ingestion_interval_ms: u64,
     controller_patterns: HashMap<QueryPatternType, Vec<PromQLPattern>>,
     query_language: QueryLanguage,
 }
@@ -147,7 +147,7 @@ impl SimpleEngine {
         // promsketch_store: Option<Arc<PromSketchStore>>,
         inference_config: InferenceConfig,
         streaming_config: Arc<StreamingConfig>,
-        prometheus_scrape_interval_ms: u64,
+        data_ingestion_interval_ms: u64,
         query_language: QueryLanguage,
     ) -> Self {
         // Create temporal pattern blocks
@@ -287,7 +287,7 @@ impl SimpleEngine {
             // promsketch_store,
             inference_config: RwLock::new(inference_config),
             streaming_config: RwLock::new(streaming_config),
-            prometheus_scrape_interval_ms,
+            data_ingestion_interval_ms,
             controller_patterns,
             query_language,
         }
@@ -332,13 +332,13 @@ impl SimpleEngine {
         mut end_timestamp: u64,
         query_pattern_type: QueryPatternType,
     ) -> u64 {
-        let interval_ms = self.prometheus_scrape_interval_ms;
+        let interval_ms = self.data_ingestion_interval_ms;
 
         if !end_timestamp.is_multiple_of(interval_ms) {
             warn!(
-                "Query end timestamp {} is not aligned with Prometheus scrape interval of {} ms. \
+                "Query end timestamp {} is not aligned with data ingestion interval of {} ms. \
                  This may lead to inaccurate results.",
-                end_timestamp, self.prometheus_scrape_interval_ms
+                end_timestamp, self.data_ingestion_interval_ms
             );
         }
 
@@ -348,8 +348,8 @@ impl SimpleEngine {
         {
             let aligned_end_timestamp = (end_timestamp / interval_ms) * interval_ms;
             debug!(
-                "OnlySpatial query: Aligning end_timestamp from {} to {} using scrape interval of {} ms",
-                end_timestamp, aligned_end_timestamp, self.prometheus_scrape_interval_ms
+                "OnlySpatial query: Aligning end_timestamp from {} to {} using data ingestion interval of {} ms",
+                end_timestamp, aligned_end_timestamp, self.data_ingestion_interval_ms
             );
             end_timestamp = aligned_end_timestamp;
         }

asap-query-engine/src/engines/simple_engine/promql.rs

@@ -43,7 +43,7 @@ impl SimpleEngine {
                     .num_milliseconds() as u64;
                 end_timestamp - range_ms
             }
-            QueryPatternType::OnlySpatial => end_timestamp - self.prometheus_scrape_interval_ms,
+            QueryPatternType::OnlySpatial => end_timestamp - self.data_ingestion_interval_ms,
         }
     }
 

asap-query-engine/src/engines/simple_engine/sql.rs

@@ -197,7 +197,7 @@ impl SimpleEngine {
                     .get_duration() as u64;
                 end_timestamp - (duration_secs * 1000)
             }
-            QueryPatternType::OnlySpatial => end_timestamp - self.prometheus_scrape_interval_ms,
+            QueryPatternType::OnlySpatial => end_timestamp - self.data_ingestion_interval_ms,
         }
     }
 
@@ -398,10 +398,10 @@ impl SimpleEngine {
             }
         };
 
-        // SQLPatternMatcher (sql_utilities, out of scope for the ms rename) divides a
+        // SQLPatternMatcher (sql_utilities, out of scope for this rename) divides a
         // seconds-denominated SQL query duration by this value — convert back to seconds.
         let matcher =
-            SQLPatternMatcher::new(schema, self.prometheus_scrape_interval_ms as f64 / 1000.0);
+            SQLPatternMatcher::new(schema, self.data_ingestion_interval_ms as f64 / 1000.0);
         let match_result = matcher.query_info_to_pattern(&query_data);
 
         debug!("Match result: {:?}", match_result);