diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 44e26adb..bb77234e 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -600,7 +600,7 @@ fn compile_physical_plan_request( exact_composition_costs: request.exact_composition_costs, erp: request.erp, planner_revision: request.planner_revision, - source_sample_interval_ms: None, + scrape_interval_ms: None, query_retention_margin_ms: 0, retained_summary_memory_budget_bytes: None, }; diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index bc6968d5..c55d2a8c 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -26,7 +26,7 @@ use planner_types::workload::{ Predictability, Query, QueryLanguage, QueryRecurrence, QueryRequirements, QueryTimeScope, QueryWorkload, Rate, RepeatedDemand, RepeatingEntry, RepetitionInterval, TimeSelection, }; -use serde::{Deserialize, Serialize}; +use serde::{de::Error as _, Deserialize, Deserializer, Serialize}; use serde_json::{json, Value}; use thiserror::Error; @@ -172,9 +172,8 @@ pub struct PhysicalCompilationRequest { /// mode falls back to theoretical sizing and then exact execution. pub erp: Option, pub planner_revision: String, - /// Observed cadence of source samples. Exact temporal panes must divide - /// both this cadence and the repeated-query evaluation interval. - pub source_sample_interval_ms: Option, + /// Source cadence used to bound current-series input lag. + pub scrape_interval_ms: Option, /// How far behind the newest ingested sample an admitted query may be /// evaluated. This extends physical retention only; it never changes the /// PromQL range selector used for readout. @@ -228,6 +227,7 @@ pub struct BackendLocalPlanningInput { /// May be absent during candidate discovery, never during deployment. #[serde(default, skip_serializing_if = "Option::is_none")] pub workload_cost_evidence: Option, + #[serde(deserialize_with = "deserialize_snapshot_query_workload")] pub query_workload: QueryWorkload, pub data_workload: DataWorkload, #[serde(rename = "implementation", alias = "physical_inputs")] @@ -243,8 +243,10 @@ pub struct BackendLocalPhysicalInputs { pub evidence_valid_for_ms: u64, pub horizon_seconds: f64, pub window_cost_model: WindowCostModel, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub source_sample_interval_ms: Option, + /// Prometheus source scrape cadence. Rangeless instant-vector expressions + /// use this as their derived readout window. This backend-local planning + /// default is distinct from Prometheus instant-selector lookback delta. + pub scrape_interval_ms: u64, #[serde(default, skip_serializing_if = "u64_is_zero")] #[serde( rename = "query_staleness_margin_ms", @@ -275,6 +277,33 @@ pub struct BackendLocalPhysicalInputs { pub erp: Option, } +/// Snapshot lookback is derived from PromQL and the declared scrape interval; +/// accepting it here would silently restore the competing legacy input. +fn deserialize_snapshot_query_workload<'de, D>(deserializer: D) -> Result +where + D: Deserializer<'de>, +{ + let value = Value::deserialize(deserializer)?; + let has_legacy_lookback = value + .get("repeating_queries") + .and_then(Value::as_array) + .is_some_and(|entries| { + entries.iter().any(|entry| { + entry + .get("time_selection") + .and_then(Value::as_object) + .and_then(|selection| selection.get("lookback")) + .is_some_and(|lookback| !lookback.is_null()) + }) + }); + if has_legacy_lookback { + return Err(D::Error::custom( + "query_workload.repeating_queries[].time_selection.lookback is not supported; it is derived from PromQL", + )); + } + serde_json::from_value(value).map_err(D::Error::custom) +} + fn u64_is_zero(value: &u64) -> bool { *value == 0 } @@ -595,23 +624,24 @@ impl BackendLocalPlanningInput { ))) } }; - let lookback_ms = entry - .time_selection - .lookback - .ok_or_else(|| { - CompileError::Snapshot(format!("query {index} requires an explicit lookback")) - })? - .0; - if lookback_ms == 0 || lookback_ms % 1_000 != 0 { - return Err(CompileError::Snapshot(format!( - "query {index} lookback must be a positive whole number of seconds" - ))); + if self.physical_inputs.scrape_interval_ms == 0 + || !self + .physical_inputs + .scrape_interval_ms + .is_multiple_of(1_000) + { + return Err(CompileError::Snapshot( + "scrape_interval_ms must be a positive whole number of seconds".into(), + )); } let accuracy = entry.requirements.accuracy.target(); let query_string = entry.query.0; let parsed = crate::query_parser::parse_query_expr_canonical(&query_string, accuracy.clone()) .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?; + let lookback_ms = + query_history_window_ms(&parsed, self.physical_inputs.scrape_interval_ms) + .map_err(|error| CompileError::Snapshot(format!("query {index}: {error}")))?; let metadata = crate::query_parser::qe_to_parsed_query(&parsed); let source_metrics = super::workload_cost::exact_source_metrics(&parsed)?; let source_hint = source_metrics.iter().next().cloned().ok_or_else(|| { @@ -695,7 +725,7 @@ impl BackendLocalPlanningInput { exact_composition_costs: exact_costs_by_id, erp: self.physical_inputs.erp, planner_revision: PLANNER_REVISION.into(), - source_sample_interval_ms: self.physical_inputs.source_sample_interval_ms, + scrape_interval_ms: Some(self.physical_inputs.scrape_interval_ms), query_retention_margin_ms: self.physical_inputs.query_retention_margin_ms, retained_summary_memory_budget_bytes: Some( self.physical_inputs.retained_summary_memory_budget_bytes, @@ -2489,12 +2519,97 @@ pub(super) fn derived_window_cost( } } +/// The furthest point in the past that evaluating `expr` can read, in ms. +/// +/// A range selector and a subquery each evaluate their child over prior +/// history; a positive offset moves that history further back. The result is a +/// lower-bound horizon, not a claim that the entire interval is read. For +/// example, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`. +fn query_history_window_ms(expr: &QueryExpr, scrape_interval_ms: u64) -> Result { + fn duration_ms(duration: std::time::Duration) -> Result { + if duration.subsec_nanos() != 0 { + return Err(CompileError::Snapshot( + "PromQL ranges must be a whole number of seconds in the backend-local profile" + .into(), + )); + } + u64::try_from(duration.as_millis()) + .map_err(|_| CompileError::Snapshot("PromQL range exceeds supported history".into())) + } + + fn add(left: u64, right: u64) -> Result { + left.checked_add(right).ok_or_else(|| { + CompileError::Snapshot("PromQL history exceeds supported duration".into()) + }) + } + + fn visit( + expr: &QueryExpr, + scrape_interval_ms: u64, + in_range: bool, + ) -> Result { + match expr { + // A subquery evaluates instant expressions at earlier timestamps; + // those leaves still need their own default readout window. + QueryExpr::PromqlSubquery { range, child, .. } => add( + duration_ms(*range)?, + visit(child, scrape_interval_ms, false)?, + ), + QueryExpr::TimeRange { range, child } => add( + duration_ms(*range)?, + visit(child, scrape_interval_ms, true)?, + ), + QueryExpr::TimeShift { shift, child } => { + if !shift.offset_ms.unsigned_abs().is_multiple_of(1_000) { + return Err(CompileError::Snapshot( + "PromQL offsets must be a whole number of seconds in the backend-local profile".into(), + )); + } + add( + shift.offset_ms.max(0) as u64, + visit(child, scrape_interval_ms, in_range)?, + ) + } + QueryExpr::PromqlScalarBridge(child) + | QueryExpr::PromqlVectorFromScalar(child) + | QueryExpr::PromqlScalarFromVector(child) + | QueryExpr::PromqlRelabel { child, .. } + | QueryExpr::PromqlSeriesSample { child, .. } + | QueryExpr::Filter { child, .. } + | QueryExpr::Project { child, .. } + | QueryExpr::Aggregate { child, .. } + | QueryExpr::Dedup { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } => visit(child, scrape_interval_ms, in_range), + QueryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } + | QueryExpr::Join { left, right, .. } + | QueryExpr::SetOp { left, right, .. } => Ok(visit( + left, + scrape_interval_ms, + in_range, + )? + .max(visit(right, scrape_interval_ms, in_range)?)), + QueryExpr::Concat { children, .. } => children.iter().try_fold(0, |history, child| { + Ok(history.max(visit(child, scrape_interval_ms, in_range)?)) + }), + QueryExpr::Scan { .. } if !in_range => Ok(scrape_interval_ms), + _ => Ok(0), + } + } + + visit(expr, scrape_interval_ms, false) +} + /// Every distinct range-selector window in `expr`, as seconds. /// /// A single query can carry several. `sum(sum_over_time(a[1m])) / sum(sum_over_time(b[5m]))` /// has two, and each one becomes its own materialization with its own window. -/// `time_selection.lookback` is the workload's declared range and is not -/// required to equal any of them. +/// These are materialization contracts. Enclosing subquery ranges and offsets +/// affect a query's furthest lookback but do not change a leaf contract. #[cfg(test)] fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet { fn visit(expr: &QueryExpr, windows: &mut BTreeSet) { @@ -2538,8 +2653,7 @@ fn range_selector_windows_secs(expr: &QueryExpr) -> BTreeSet { } /// Derive and price one implementation per supported layout for each range. -/// Explicit snapshot candidates bypass this path. After logical selection, -/// derived maintenance cohorts are restricted to their supported full windows; +/// Derived maintenance cohorts are restricted to their supported full windows; /// raw additive pane producers may subsequently be shared by Planner. #[cfg(test)] fn derived_window_candidates( @@ -3755,6 +3869,10 @@ pub(crate) mod tests { .. } = node { + // Scrape cadence bounds input lag; selector staleness keeps + // the independently declared Planner lookback. + assert_eq!(population.max_input_lag_ms, 5_000); + assert_eq!(population.lookback_ms, 300_000); assert_eq!(population.max_k, 5); assert!(population.quantiles); populations.insert(population.key()); @@ -4018,7 +4136,6 @@ pub(crate) mod tests { let mut snapshot = planning_snapshot(); let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; entry.query = Query(text.into()); - entry.time_selection.lookback = Some(DurationMs(300_000)); if !text.contains("quantile") { entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); } @@ -4625,7 +4742,7 @@ pub(crate) mod tests { erp: None, exact_composition_costs: HashMap::new(), planner_revision: PLANNER_REVISION.into(), - source_sample_interval_ms: None, + scrape_interval_ms: None, query_retention_margin_ms: 0, retained_summary_memory_budget_bytes: None, }) @@ -6049,6 +6166,91 @@ pub(crate) mod tests { .unwrap() } + fn derived_query_window_secs(query: &str) -> u64 { + let mut snapshot = planning_snapshot(); + let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; + entry.query = Query(query.into()); + entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact); + snapshot + .into_physical_compilation_request() + .unwrap() + .0 + .queries[0] + .query_lookback_seconds + } + + // Every concatenated histogram result contributes its source history. + #[test] + fn derived_history_visits_concat_branches() { + assert_eq!(derived_query_window_secs( + "histogram_quantiles(rate(request_duration_seconds_bucket[5m]), \"quantile\", 0.5, 0.9)" + ), 300); + } + + // A short ranged sibling must not suppress a raw selector's default window. + #[test] + fn derived_history_applies_cadence_at_each_rangeless_source() { + assert_eq!( + derived_query_window_secs("sum(a) + sum(sum_over_time(b[1s]))"), + 5 + ); + assert_eq!(derived_query_window_secs("sum(a offset 1h)"), 3605); + assert_eq!(derived_query_window_secs("sum_over_time(a[1s])"), 1); + } + + // The seconds-based backend must fail explicitly instead of shrinking history. + #[test] + fn derived_history_rejects_fractional_seconds() { + for query in [ + "sum_over_time(a[1500ms])", + "sum_over_time(a[500ms])", + "sum_over_time(a[1500ms] offset 500ms)", + "sum_over_time(a[1s]) + sum(sum_over_time(b[500ms]))", + "avg_over_time((sum(a))[1500ms:])", + "sum(a offset 500ms)", + ] { + let mut snapshot = planning_snapshot(); + snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query = + Query(query.into()); + let error = snapshot + .into_physical_compilation_request() + .expect_err(query) + .to_string(); + assert!( + error.contains("whole number of seconds"), + "{query}: {error}" + ); + } + } + + #[test] + fn snapshot_rejects_manual_lookback() { + let mut wire = serde_json::to_value(planning_snapshot()).unwrap(); + wire["query_workload"]["repeating_queries"][0]["time_selection"]["lookback"] = + 60_000.into(); + let error = serde_json::from_value::(wire) + .expect_err("manual lookback must not be accepted") + .to_string(); + assert!(error.contains("lookback is not supported"), "{error}"); + } + + #[test] + fn subquery_range_derives_the_query_window() { + assert_eq!( + derived_query_window_secs("avg_over_time((sum(a))[6h:])"), + 6 * 60 * 60 + 5 + ); + } + + #[test] + fn offset_extends_the_query_furthest_lookback() { + // Evaluated at t, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`. + assert_eq!( + derived_query_window_secs("sum_over_time(a[1m] offset 1h)"), + 61 * 60 + ); + } + // A workload evaluated more often than its window is wide must advance its // state at that cadence. Planning it as one lookback-wide tumbling window // answers with results that only change once per window. @@ -6453,7 +6655,6 @@ pub(crate) mod tests { { let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0]; entry.query = Query("quantile_over_time(0.5, data[5m])".into()); - entry.time_selection.lookback = Some(DurationMs(300_000)); entry.demand = RepeatedDemand::FixedIntervalAt { interval: RepetitionInterval(30_000), evaluation_phase: planner_types::workload::TimestampMs(0), @@ -7001,7 +7202,7 @@ pub(crate) mod tests { predictability: Predictability::Predictable { known_at: None }, time_selection: TimeSelection { scope: QueryTimeScope::RealTime, - lookback: Some(DurationMs(60_000)), + lookback: None, as_of: None, }, }]), @@ -7028,7 +7229,7 @@ pub(crate) mod tests { cost: template.window_realization_candidates[0].cost.clone(), quotes: Vec::new(), }, - source_sample_interval_ms: None, + scrape_interval_ms: 1_000, query_retention_margin_ms: 0, retained_summary_memory_budget_bytes: DEFAULT_RETAINED_SUMMARY_MEMORY_BUDGET_BYTES, topk_evidence: HashMap::new(), diff --git a/control_plane/src/physical/maintained_population.rs b/control_plane/src/physical/maintained_population.rs index efd045cd..2b1270b1 100644 --- a/control_plane/src/physical/maintained_population.rs +++ b/control_plane/src/physical/maintained_population.rs @@ -89,7 +89,7 @@ pub(super) fn operator( max_bytes, max_series: 100_000.min((max_bytes / 1024) as usize), max_input_lag_ms: request - .source_sample_interval_ms + .scrape_interval_ms .unwrap_or(60_000) .saturating_add(request.query_retention_margin_ms) .clamp(1, 300_000), diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index eb535f5d..8b5eff65 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -1101,7 +1101,6 @@ mod tests { entries[0].requirements.accuracy = planner_types::workload::AccuracyRequirement::Explicit( planner_types::types::AccuracyTarget::Exact, ); - entries[0].time_selection.lookback = Some(planner_types::workload::DurationMs(21_600_000)); let mut second = entries[0].clone(); second.query = planner_types::workload::Query( "max_over_time(service_retry_queue_depth{job=\"order-service\"}[6h])".into(), diff --git a/control_plane/src/query_plan/residual.rs b/control_plane/src/query_plan/residual.rs index 05845755..cd3f5c98 100644 --- a/control_plane/src/query_plan/residual.rs +++ b/control_plane/src/query_plan/residual.rs @@ -575,19 +575,6 @@ mod planner_workload_tests { use super::*; use crate::physical::compiler::{BackendLocalPlanningInput, PhysicalPlanCompiler}; - fn lookback(expr: &Expr) -> u64 { - match expr { - Expr::MatrixSelector(e) => e.range.as_millis() as u64, - Expr::Subquery(e) => (e.range.as_millis() as u64).max(lookback(&e.expr)), - Expr::Aggregate(e) => lookback(&e.expr), - Expr::Paren(e) => lookback(&e.expr), - Expr::Unary(e) => lookback(&e.expr), - Expr::Binary(e) => lookback(&e.lhs).max(lookback(&e.rhs)), - Expr::Call(e) => e.args.args.iter().map(|e| lookback(e)).max().unwrap_or(0), - _ => 0, - } - } - fn compile_one(query: &str) -> crate::physical::compiler::CompiledPhysicalPlan { let mut fixture: serde_json::Value = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-planning-snapshot.json" @@ -596,8 +583,6 @@ mod planner_workload_tests { let mut entry = fixture["query_workload"]["repeating_queries"][0].clone(); entry["query"] = query.into(); entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"}); - let window = lookback(&parser::parse(query).unwrap()); - entry["time_selection"]["lookback"] = (if window == 0 { 300_000 } else { window }).into(); fixture["query_workload"]["repeating_queries"] = vec![entry].into(); let snapshot: BackendLocalPlanningInput = serde_json::from_value(fixture).unwrap(); let (request, environment) = snapshot @@ -683,9 +668,6 @@ mod planner_workload_tests { let mut entry = template.clone(); entry["query"] = query.into(); entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"}); - let window = lookback(&parser::parse(query).unwrap()); - entry["time_selection"]["lookback"] = - (if window == 0 { 300_000 } else { window }).into(); entries.push(entry); } fixture["query_workload"]["repeating_queries"] = entries.into(); diff --git a/control_plane/tests/discovery_snapshot.rs b/control_plane/tests/discovery_snapshot.rs new file mode 100644 index 00000000..03b88164 --- /dev/null +++ b/control_plane/tests/discovery_snapshot.rs @@ -0,0 +1,58 @@ +//! Discovery output must be accepted by the production snapshot planner. +use control_plane::physical::compiler::BackendLocalPlanningInput; +use std::{fs, path::PathBuf, process::Command}; + +struct TempDirectory(PathBuf); +impl Drop for TempDirectory { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +#[test] +fn discovered_snapshot_plans_with_observed_cadence_and_promql_history() { + let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .unwrap() + .to_path_buf(); + let temporary = TempDirectory( + std::env::temp_dir().join(format!("asap-discovery-snapshot-{}", std::process::id())), + ); + fs::create_dir(&temporary.0).unwrap(); + let corpus = temporary.0.join("corpus.json"); + let metrics = temporary.0.join("metrics.prom"); + let output = temporary.0.join("snapshot.json"); + fs::write(&corpus, r#"{"upstream_revision":"test","queries":[{"id":"q","query":"max_over_time(m[1m] offset 1h)","eval_timestamp_ms":120000}]}"#).unwrap(); + fs::write(&metrics, "m{job=\"test\"} 1 60\nm{job=\"test\"} 2 120\n").unwrap(); + let result = Command::new("python3") + .env("PYTHONDONTWRITEBYTECODE", "1") + .arg(root.join("tools/o11y-execution/discover_snapshot.py")) + .arg("--corpus") + .arg(&corpus) + .arg("--metrics") + .arg(&metrics) + .arg("--template") + .arg(root.join("docs/examples/asapquery-planning-snapshot.json")) + .arg("--output") + .arg(&output) + .args(["--repetitions", "1"]) + .output() + .unwrap(); + assert!( + result.status.success(), + "{}", + String::from_utf8_lossy(&result.stderr) + ); + let snapshot: BackendLocalPlanningInput = + serde_json::from_str(&fs::read_to_string(&output).unwrap()).unwrap(); + assert_eq!(snapshot.physical_inputs.scrape_interval_ms, 60_000); + let (request, _) = snapshot + .clone() + .into_physical_compilation_request() + .unwrap(); + assert_eq!(request.scrape_interval_ms, Some(60_000)); + assert_eq!(request.queries[0].query_lookback_seconds, 3660); + let roundtrip: BackendLocalPlanningInput = + serde_json::from_value(serde_json::to_value(&snapshot).unwrap()).unwrap(); + assert_eq!(snapshot, roundtrip); +} diff --git a/data_plane/src/precompute_engine/multisource_coordinator.rs b/data_plane/src/precompute_engine/multisource_coordinator.rs index 2e3dcc5f..d429d774 100644 --- a/data_plane/src/precompute_engine/multisource_coordinator.rs +++ b/data_plane/src/precompute_engine/multisource_coordinator.rs @@ -615,7 +615,6 @@ mod tests { query["query"] = "quantile(0.9, sum_over_time(m[1m]) + sum_over_time(n[1m]))".into(); query["demand"]["fixed_interval_at"]["interval"] = 60000.into(); query["demand"]["fixed_interval_at"]["evaluation_phase"] = 0.into(); - query["time_selection"]["lookback"] = 60000.into(); wire["query_workload"]["repeating_queries"] = serde_json::json!([query]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningInput = serde_json::from_value(wire).unwrap(); diff --git a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs index 65c5907b..c1bc5f22 100644 --- a/data_plane/src/storage_engines/sketch_db/index/maintenance.rs +++ b/data_plane/src/storage_engines/sketch_db/index/maintenance.rs @@ -1082,7 +1082,6 @@ mod tests { entry["query"] = "quantile(0.9, sum_over_time(immutable_value[1m]))".into(); entry["demand"]["fixed_interval_at"]["interval"] = 60_000.into(); entry["demand"]["fixed_interval_at"]["evaluation_phase"] = 0.into(); - entry["time_selection"]["lookback"] = 60_000.into(); fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningInput = serde_json::from_value(fixture).unwrap(); diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index ec5f729f..51df1d98 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -801,9 +801,6 @@ async fn run_shared_dashboard(multi_pane: bool) { let mut typed: control_plane::physical::compiler::BackendLocalPlanningInput = serde_json::from_value(snapshot.clone()).unwrap(); if multi_pane { - for entry in typed.query_workload.repeating_queries.as_mut().unwrap() { - entry.time_selection.lookback = Some(planner_types::workload::DurationMs(10_000)); - } for entry in typed.query_workload.repeating_queries.as_mut().unwrap() { entry.demand = planner_types::workload::RepeatedDemand::FixedIntervalAt { interval: planner_types::workload::RepetitionInterval(5_000), diff --git a/data_plane/tests/support/current_series_process.rs b/data_plane/tests/support/current_series_process.rs index cd383322..b4e36951 100644 --- a/data_plane/tests/support/current_series_process.rs +++ b/data_plane/tests/support/current_series_process.rs @@ -26,7 +26,7 @@ async fn current_series_quantiles_topk_share_and_replace_values() { )) .unwrap(); snapshot.schema_version = 2; - snapshot.physical_inputs.source_sample_interval_ms = Some(60_000); + snapshot.physical_inputs.scrape_interval_ms = 60_000; let template = snapshot.query_workload.repeating_queries.as_ref().unwrap()[0].clone(); let mut queries = vec![]; for q in [0.5, 0.9, 0.95, 0.99] { diff --git a/data_plane/tests/support/immutable_maintenance_process.rs b/data_plane/tests/support/immutable_maintenance_process.rs index a1bbf43f..eeeee37d 100644 --- a/data_plane/tests/support/immutable_maintenance_process.rs +++ b/data_plane/tests/support/immutable_maintenance_process.rs @@ -31,7 +31,6 @@ async fn run_maintenance_process(multi_source: bool, distinct_groups: bool) { entry["query"] = query.into(); entry["demand"]["fixed_interval_at"]["interval"] = 60_000.into(); entry["demand"]["fixed_interval_at"]["evaluation_phase"] = 0.into(); - entry["time_selection"]["lookback"] = 60_000.into(); fixture["query_workload"]["repeating_queries"] = serde_json::json!([entry]); let snapshot: control_plane::physical::compiler::BackendLocalPlanningInput = serde_json::from_value(fixture.clone()).unwrap(); diff --git a/data_plane/tests/support/issue_701_702_process.rs b/data_plane/tests/support/issue_701_702_process.rs index 9a0420f5..4baff6d4 100644 --- a/data_plane/tests/support/issue_701_702_process.rs +++ b/data_plane/tests/support/issue_701_702_process.rs @@ -121,15 +121,14 @@ async fn issue_workloads_execute_warm_at_successive_evaluations() { "../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - fixture["implementation"]["source_sample_interval_ms"] = 1000.into(); + fixture["implementation"]["scrape_interval_ms"] = 1000.into(); fixture["implementation"]["horizon_seconds"] = 3600.into(); let template = fixture["query_workload"]["repeating_queries"][0].clone(); fixture["query_workload"]["repeating_queries"] = queries .iter() - .map(|(query, lookback, cadence)| { + .map(|(query, _lookback, cadence)| { let mut entry = template.clone(); entry["query"] = query.clone().into(); - entry["time_selection"]["lookback"] = (lookback * 1000).into(); entry["demand"]["fixed_interval_at"]["interval"] = (cadence * 1000).into(); if !query.contains("quantile") { entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"}); @@ -359,13 +358,12 @@ async fn temporal_average_overflow_falls_back_after_state_is_warm() { "../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); - fixture["implementation"]["source_sample_interval_ms"] = 1000.into(); + fixture["implementation"]["scrape_interval_ms"] = 1000.into(); let template = fixture["query_workload"]["repeating_queries"][0].clone(); fixture["query_workload"]["repeating_queries"] = ["avg", "sum", "count"] .map(|op| { let mut entry = template.clone(); entry["query"] = format!("{op}_over_time(average_overflow[5s])").into(); - entry["time_selection"]["lookback"] = 5000.into(); entry["demand"]["fixed_interval_at"]["interval"] = 1000.into(); entry["requirements"]["accuracy"] = serde_json::json!({"explicit":"Exact"}); entry diff --git a/docs/developer_docs/planning/repeated-dashboard-panes.md b/docs/developer_docs/planning/repeated-dashboard-panes.md index 69bef94c..119c2014 100644 --- a/docs/developer_docs/planning/repeated-dashboard-panes.md +++ b/docs/developer_docs/planning/repeated-dashboard-panes.md @@ -25,6 +25,28 @@ Unquoted layouts use supplied lifecycle unit costs multiplied by structural counts. Layout changes never inherit another layout's measured scalar cost. Workload-versus-exact deployment evidence is still required by snapshot version 2. +Temporal requirements are derived from PromQL. A range selector supplies its +own readout window; each rangeless source uses required +`implementation.scrape_interval_ms` before enclosing offsets and subqueries +are added. Branches, including concatenations, contribute their maximum history. +For example, with a 5-second cadence, `sum(a offset 1h)` needs 3605 seconds, +and `avg_over_time((sum(a))[6h:])` needs 21605 seconds. A ranged source keeps +its explicit window even when it is shorter than the scrape cadence. + +This is the backend-local cadence-based planning contract, not Prometheus's +[instant-selector lookback delta](https://prometheus.io/docs/prometheus/latest/querying/basics/#staleness) +(which defaults to five minutes and governs selection of the latest non-stale sample). Scrape cadence does not configure +that Prometheus setting or provide its staleness semantics. + +Non-null snapshot `time_selection.lookback` is rejected because it duplicates +and can conflict with query semantics; omission and `null` are accepted. +The backend stores window widths in seconds, so fractional-second ranges and +offsets are rejected explicitly rather than rounded down. + +A subquery or positive offset extends only the furthest lookback: evaluated at +`t`, `a[1m] offset 1h` selects `(t - 61m, t - 60m]`, not a continuous +61-minute interval. + ## Cases All ranges below use PromQL's `(start,end]` convention. W is the lookback and E diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index 87ecd811..0c466eff 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -9,21 +9,21 @@ "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } }, "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" }, "predictability": { "predictable": { "known_at": null } }, - "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + "time_selection": { "scope": "real_time", "lookback": null, "as_of": null } }, { "query": "increase(asap_demo_counter_total[5s])", "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } }, "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" }, "predictability": { "predictable": { "known_at": null } }, - "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + "time_selection": { "scope": "real_time", "lookback": null, "as_of": null } }, { "query": "sum(sum_over_time(asap_demo_gauge[5s]))", "demand": { "fixed_interval_at": { "interval": 1000, "evaluation_phase": 0 } }, "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" }, "predictability": { "predictable": { "known_at": null } }, - "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + "time_selection": { "scope": "real_time", "lookback": null, "as_of": null } }, { "query": "quantile_over_time(0.5, asap_demo_latency_ms[5s])", @@ -33,7 +33,7 @@ "response_latency": "unspecified" }, "predictability": { "predictable": { "known_at": null } }, - "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + "time_selection": { "scope": "real_time", "lookback": null, "as_of": null } }, { "query": "topk(1, sum_over_time(asap_demo_gauge[5s]))", @@ -43,7 +43,7 @@ "response_latency": "unspecified" }, "predictability": { "predictable": { "known_at": null } }, - "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + "time_selection": { "scope": "real_time", "lookback": null, "as_of": null } }, { "query": "topk(1, count_over_time(asap_demo_gauge[5s]))", @@ -53,7 +53,7 @@ "response_latency": "unspecified" }, "predictability": { "predictable": { "known_at": null } }, - "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + "time_selection": { "scope": "real_time", "lookback": null, "as_of": null } } ], "data_workload": { @@ -98,6 +98,7 @@ "weighted_cost": 1.0 } }, + "scrape_interval_ms": 5000, "topk_evidence": { "topk(1, sum_over_time(asap_demo_gauge[5s]))": { "selected_lower_bound": 101.0, diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json index b4147ae6..4f0ed030 100644 --- a/docs/examples/asapquery-planning-snapshot.json +++ b/docs/examples/asapquery-planning-snapshot.json @@ -18,7 +18,7 @@ "predictability": { "predictable": { "known_at": null } }, "time_selection": { "scope": "real_time", - "lookback": 60000, + "lookback": null, "as_of": null } } @@ -64,7 +64,8 @@ "source_scan_bytes": 0, "weighted_cost": 1.0 } - } + }, + "scrape_interval_ms": 5000 }, "environment": { "target": "backend_local_remote_write", diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md index 1cf44766..fba4e022 100644 --- a/docs/user_guide/asapquery-profile.md +++ b/docs/user_guide/asapquery-profile.md @@ -137,7 +137,12 @@ coverage is a capability miss, never a partial warm success. Inspect the per-materialization state and observed coverage through `GET /api/v1/physical-plan/status`. -Snapshot schema version `1` currently accepts fixed-interval repeating PromQL -queries with explicit whole-second lookbacks and fresh ingestion-rate -evidence. Unsupported snapshot semantics fail startup rather than silently -inventing cost or placement evidence. +Snapshot schema version `2` accepts fixed-interval repeating PromQL queries +with `implementation.scrape_interval_ms` and fresh ingestion-rate evidence. +Range selectors derive their own lookback; each rangeless source uses the +scrape interval for backend-local planning. This default window is distinct +from Prometheus's instant-selector lookback delta and does not configure it. +Omit `time_selection.lookback` or set it to `null`. Ranges and offsets must be +whole seconds; unsupported fractional durations fail startup instead of +silently shortening the window. Unsupported snapshot semantics fail startup +rather than silently inventing cost or placement evidence. diff --git a/tools/o11y-execution/discover_snapshot.py b/tools/o11y-execution/discover_snapshot.py index ad72510c..7308b925 100644 --- a/tools/o11y-execution/discover_snapshot.py +++ b/tools/o11y-execution/discover_snapshot.py @@ -10,17 +10,11 @@ import hashlib import json from pathlib import Path -import re import time import math from replay import iter_samples -def duration_ms(text): - units = {"ms": 1, "s": 1000, "m": 60000, "h": 3600000, "d": 86400000, "w": 604800000, "y": 31536000000} - return sum(int(n) * units[u] for n, u in re.findall(r"(\d+)(ms|[smhdwy])", text)) - - def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--corpus", type=Path, required=True) @@ -67,8 +61,6 @@ def evidence(value): registrations, query_audit = [], [] frequencies = Counter(row["query"] for row in corpus["queries"]) for query in dict.fromkeys(row["query"] for row in corpus["queries"]): - windows = [duration_ms(x) for x in re.findall(r"\[([0-9a-z]+)(?::[^\]]*)?\]", query)] - lookback = max(windows, default=300000) if args.interval_ms % frequencies[query]: raise ValueError("base interval must divide exactly by query occurrence frequency") interval = args.interval_ms // frequencies[query] @@ -79,17 +71,22 @@ def evidence(value): registrations.append({"query": query, "demand": {"fixed_interval_at": {"interval": interval, "evaluation_phase": phase}}, "requirements": {"accuracy": {"explicit": "Exact"}, "response_latency": "unspecified"}, "predictability": {"predictable": {"known_at": None}}, - "time_selection": {"scope": "real_time", "lookback": lookback, "as_of": None}}) - query_audit.append({"query": query, "window_lookback_ms": lookback, + "time_selection": {"scope": "real_time", "lookback": None, "as_of": None}}) + query_audit.append({"query": query, "occurrence_count": frequencies[query], "expected_evaluations": frequencies[query] * args.repetitions, "declared_interval_ms": interval, "evaluation_phase_ms": phase, - "lookback_method": "largest explicit range; instant selector defaults to Prometheus 5m; original offsets/subqueries preserved in query"}) + "lookback_method": "derived by the backend compiler from PromQL and the declared scrape cadence"}) snapshot["query_workload"].update(repeating_queries=registrations, data_workload=data, query_batch=None) snapshot["snapshot_version"] = 2 snapshot.pop("workload_cost_evidence", None) implementation = snapshot["implementation"] - if source_sample_interval_ms: - implementation["source_sample_interval_ms"] = source_sample_interval_ms + implementation.pop("source_sample_interval_ms", None) + # Prefer the observed cadence; retain the declared template cadence when + # the input has no repeated series from which to infer it. + scrape_interval_ms = source_sample_interval_ms or implementation.get("scrape_interval_ms") + if not isinstance(scrape_interval_ms, int) or scrape_interval_ms <= 0 or scrape_interval_ms % 1000: + raise ValueError("scrape_interval_ms must be a positive whole number of seconds") + implementation["scrape_interval_ms"] = scrape_interval_ms # Finite replay evaluates the oldest repetition first after loading the # complete input. Preserve that admitted historical-query span separately # from each query's PromQL range selector. diff --git a/tools/o11y-execution/test_discover_snapshot.py b/tools/o11y-execution/test_discover_snapshot.py index 94497ec8..f195eb77 100644 --- a/tools/o11y-execution/test_discover_snapshot.py +++ b/tools/o11y-execution/test_discover_snapshot.py @@ -25,6 +25,10 @@ def test_historical_input_does_not_backdate_plan_activation(self): ], check=True) snapshot = json.loads(output.read_text()) self.assertEqual(snapshot["query_workload"]["repeating_queries"][0]["demand"], {"fixed_interval_at": {"interval": 60000, "evaluation_phase": 0}}) + # Discovery output must obey the same schema as the compiler input. + self.assertIsNone(snapshot["query_workload"]["repeating_queries"][0]["time_selection"].get("lookback")) + self.assertEqual(snapshot["implementation"]["scrape_interval_ms"], 60_000) + self.assertNotIn("source_sample_interval_ms", snapshot["implementation"]) environment = snapshot["environment"] self.assertEqual(environment["activation_unix_ms"], environment["observed_at_unix_ms"]) provenance = json.loads(output.with_suffix(".provenance.json").read_text())