diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md new file mode 100644 index 00000000..58a0764e --- /dev/null +++ b/.design_docs/cost-optimizer-decision-map.md @@ -0,0 +1,147 @@ +# Cost-based optimizer decision map + +Goal: turn a PromQL workload with query frequencies and dataset-conditioned +sketch measurements into an offline deployment plan, then collect evidence that +the selected plan improves the stated objective without violating accuracy. + +## Progress log + +- 2026-09-07: Located the existing offline planner entry point: + `asap-optimizer-cli`. `Controller::generate()` still takes the hardcoded + planner path, so production wiring is intentionally not the first milestone. +- 2026-09-07: Verified a wire-contract gap: sketch-bench emits a versioned, + workload-profiled atomic-cost document, while ASAPQuery currently loads a + legacy flat entry array. A profile-aware loader and explicit selector are the + first integration change. +- 2026-09-07: Distinguished grouping labels from sketch keys. Grouping labels + determine how many sketch instances the planner deploys; sketch-key + properties belong to a measured benchmark profile. +- 2026-09-07: Chosen first vertical slice: KLL backing PromQL + `quantile_over_time`. The next unresolved concrete inputs are a source + metric, KLL input value/key column, trace time slice, and grouping labels. +- 2026-09-07: Candidate external workload specs already present in + `sketch-bench`: Alibaba microservices CPU (`cpu_utilization`, grouped by + `msname`, one-minute slice); Google task CPU (`cpu_rate`, grouped by + `machine_id`, three-minute slice); and Datadog/BOOM (`target`, scalar, + one-minute slice). Existing ASAPQuery Prometheus replay configurations also + exist for Alibaba node CPU and Google CPU. Recommended first case: Google + CPU, because it has both a sketch-bench workload spec and a matching + Prometheus replay configuration with explicit labels. +- 2026-09-07: Verified the selected Google trace is available locally: + `google-cluster-data/ClusterData2011/clusterdata-2011-2/task_usage/part-00262-of-00500.csv.gz` + (92 MB). The external loader supports numeric f64 grouped workloads, and the + selected spec uses `cpu_rate`, grouped by `machine_id`, over a three-minute + time slice. `kll-percall` is the correct benchmark variant because the + ASAPQuery runtime invokes KLL `quantile()` per query. Remaining prerequisites: + (1) a matching `quantile_over_time` workload and series-inventory CSV, + (2) profile-aware atomic-cost loading/selection in ASAPQuery, and + (3) explicit accuracy, arrival-rate, and exact-baseline assumptions. +- 2026-09-07: Inspected sketch-bench PR #124. It is the correct producer-side + contract: a versioned document, one profile per exact external workload, + required `query_accuracy`, and no cross-profile merging. Decision: do not + add an ASAPQuery-specific `logical_metric` field to PR #124. Its + `value_column` identifies the physical trace column measured; a Prometheus + metric name is a query-engine identity that may rename or transform that + column. ASAPQuery owns the explicit mapping from a queried metric to a + benchmark profile. Do not add KLL key columns for this slice: KLL consumes + numeric values and keyed external workloads are intentionally unsupported. + ASAPQuery should mirror the versioned types, select exactly one external + profile from an explicit selector, and pass only its entries to the existing + candidate resolver. +- 2026-09-07: Corrected an earlier naming example after inspecting collector + code. `google_mean_cpu_usage_rate_0` is an older experiment-specific name; + the existing evaluation mapper emits `google_cluster_2019_cpu_rate` from + raw `cpu_rate`, and creates a `_q_kll` alias only to route a second copy of + the same samples through KLL. These are naming/routing conventions, not + distinct source values. For the first slice, use one canonical raw-stream + identity throughout (recommended Prometheus-safe name: + `google_task_usage_cpu_rate`), and do not require a separate logical-metric + field in the sketch-bench document. +- 2026-09-07: Validated the PR-123/PR-124 dependency analysis. PR 123 changes + the underlying benchmark-record schema; PR 124 is the atomic-cost consumer + break and must be handled via its independent document schema version. Two + implementation corrections: `query_accuracy` is a required map of named + metrics, not one `f64`; and workload selection can happen once at the + ASAPQuery load boundary, returning the selected profile's existing flat + `AtomicCostTable`. This avoids threading workload identity through every + candidate resolver while still rejecting zero or ambiguous matches before + any candidate lookup. +- 2026-09-07: Implemented the ASAPQuery consumer seam on branch + `feat/profiled-atomic-cost-loader`. It mirrors PR 124's versioned document, + external workload, required accuracy map, strict schema validation, and + exact single-profile selection. The two offline optimizer CLIs now require a + JSON `profiles[].workload` selector whenever `--atomic-costs` is supplied; + selected entries preserve the existing flat resolver interface. Focused + loader and full planner tests pass (216 tests plus doc-tests); the latter was + run outside the sandbox because its existing ClickHouse mock opens a local + listener. +- 2026-09-07: Review follow-up: centralized CLI profile loading behind one + optimizer module interface and added explicit zero-match coverage. Synthetic + profile descriptions remain opaque JSON because sketch-bench's data-generator + schema is independently versioned; document, profile, entry, and external + workload fields remain strict at this consumer boundary. + +## #1: What identifies a benchmark cost profile? + +Type: Discuss + +### Question + +Which dataset properties are expected to change insert, merge, query, memory, +or accuracy enough that they must select a distinct atomic-cost profile? Decide +the first paper-scale profile matrix and the semantics for choosing a profile at +planning time. + +### Answer + +Open. `sketch-bench` already emits `AtomicCostDocument { schema_version, +profiles: [{ workload, entries }] }`; the optimizer currently reads the older +flat entry array and therefore has no profile-selection rule. The available +observability corpus is under `../benchmarks/metrics_observability/data` from +the workspace root (Datadog/BOOM and Alibaba traces). Candidate framing from +the user: `(dataset_name, metric_name, keying/aggregated-label names, +time-range)`. The remaining decision is whether labels and range identify an +atomic measurement or instead parameterize the structural cost formula. + +Clarification: **grouping labels** are the PromQL `GROUP BY` labels and +partition the metric stream into separate sketch instances. A **sketch key** +is the value or label tuple inserted into keyed sketches such as HLL, CMS, and +Hydra. Grouping labels and their observed distinct-group count are planner +context; the sketch-key distribution/cardinality (and, where relevant, encoded +key size) belong in the benchmark profile. + +For the first vertical slice, scope this decision to one KLL +`quantile_over_time` workload. Do not define the full cross-dataset matrix yet. + +## #2: What is the minimum credible empirical planning loop? + +Blocked by: #1 +Type: Prototype + +### Question + +What end-to-end experiment should prove that measured costs, rather than +hand-tuned constants, change a planner decision appropriately for a fixed +PromQL workload and series inventory? + +### Answer + +Open. The existing offline `asap-optimizer-cli` is the intended harness after +the document-loader/profile-selection gap is closed. + +## #3: What feasibility evidence constrains optimization? + +Blocked by: #1, #2 +Type: Discuss + +### Question + +For each query/sketch/config/dataset case, which accuracy metric and threshold +make a candidate eligible, and how will exact-query cost and arrival rate be +measured rather than assumed? + +### Answer + +Open. `sketch-bench` already retains capability-specific `query_accuracy`, but +the ASAPQuery greedy optimizer does not use it; arrival rate (`rho`) and exact +query cost are currently placeholders. diff --git a/.design_docs/optimizer-v1-implementation-plan.md b/.design_docs/optimizer-v1-implementation-plan.md index 2fde701f..658b172e 100644 --- a/.design_docs/optimizer-v1-implementation-plan.md +++ b/.design_docs/optimizer-v1-implementation-plan.md @@ -322,7 +322,8 @@ cargo run -p asap_planner --bin asap-optimizer-cli -- \ --dataset \ --data-ingestion-interval-ms 60000 \ [--rho 1.0] \ - [--atomic-costs ] + [--atomic-costs \ + --atomic-cost-workload ] ``` Takes the same `ControllerConfig` YAML format as `asap-planner --input_config`. @@ -332,7 +333,11 @@ group count; no live Prometheus connection is needed. A `metrics:` hints block, present, is checked against the dataset and mismatches fail loudly. Prints deployed streaming configs and query configs to stdout. `--rho` is the placeholder arrival rate (see TODOs below — not real yet). `--atomic-costs` is -optional; omit it and ordinary unbenchmarked candidates use the flat stub, while +optional; when supplied it requires `--atomic-cost-workload`, a JSON file +containing the exact `profiles[].workload` value from that benchmark artifact. +The loader validates the document schema and rejects a selector that matches +zero or multiple profiles; it never mixes entries across workloads. Omit both +flags and ordinary unbenchmarked candidates use the flat stub, while CMS-with-heap candidates warn and are dropped until a matching reference row is available. ### Running with real sketch-bench costs @@ -352,13 +357,15 @@ CMS-with-heap candidates warn and are dropped until a matching reference row is # See what each candidate would cost, before selection: cargo run -p asap_planner --bin candidate-gen-dump -- \ --input_config workload.yaml --data-ingestion-interval-ms 60000 \ - --atomic-costs path/to/atomic_costs.json + --atomic-costs path/to/atomic_costs.json \ + --atomic-cost-workload path/to/workload-selector.json # Run the actual optimizer: cargo run -p asap_planner --bin asap-optimizer-cli -- \ --input_config workload.yaml --data-ingestion-interval-ms 60000 \ --dataset path/to/series-inventory.csv \ - --atomic-costs path/to/atomic_costs.json + --atomic-costs path/to/atomic_costs.json \ + --atomic-cost-workload path/to/workload-selector.json ``` `candidate-gen-dump`'s output labels each resolved params row `[real]` or `[stub]`; candidates diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 00000000..a83c52ec --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,24 @@ +# ASAPQuery + +Terms used by the cost-based sketch optimizer and its benchmarking pipeline. + +## Language + +**Grouping labels**: +PromQL `GROUP BY` labels that partition a metric stream into label groups; a +non-subpopulation sketch is instantiated once per distinct group. +_Avoid_: keying labels + +**Sketch key**: +The value or label tuple inserted into and queried from a keyed sketch such as +HLL, CMS, or Hydra. Its distribution, cardinality, and encoded size may affect +the measured atomic cost and accuracy. +_Avoid_: grouping label, partition key + +**Benchmark profile**: +A reproducible description of the input trace slice used to measure an +atomic-cost table for a sketch family and configuration. + +**Planner context**: +Query- and deployment-specific structural inputs, such as label-group count +and retained-window count, that scale atomic costs into a plan cost. diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs index a900125f..b0af6864 100644 --- a/asap-planner-rs/src/bin/candidate_gen_dump.rs +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -6,8 +6,8 @@ use std::path::PathBuf; use asap_planner::{ optimizer::{ - enumerate_candidates, extract_aqes, load_atomic_cost_table, resolve_atomic_costs, - AtomicCostTable, AtomicCosts, CandidateConfig, RQE, + enumerate_candidates, extract_aqes, load_optional_selected_atomic_cost_table, + resolve_atomic_costs, AtomicCostTable, AtomicCosts, CandidateConfig, RQE, }, ControllerConfig, }; @@ -28,21 +28,26 @@ struct Args { #[arg(long = "data-ingestion-interval-ms")] scrape_interval_ms: u64, - /// Path to sketch-bench's exported atomic-cost table (see ASAPQuery#524). + /// Path to sketch-bench's versioned atomic-cost document. /// When given, each params row also prints its resolved AtomicCosts -- /// real (from the table) or the flat stub (unbenchmarked family, or this /// exact param point missing from the table) -- labeled which. #[arg(long = "atomic-costs")] atomic_costs: Option, + + /// JSON `profiles[].workload` value selecting exactly one measured profile. + #[arg(long = "atomic-cost-workload", requires = "atomic_costs")] + atomic_cost_workload: Option, } fn main() -> anyhow::Result<()> { let args = Args::parse(); - let atomic_cost_table = match &args.atomic_costs { - Some(path) => load_atomic_cost_table(path)?, - None => AtomicCostTable::default(), - }; + let atomic_cost_table = load_optional_selected_atomic_cost_table( + args.atomic_costs.as_deref(), + args.atomic_cost_workload.as_deref(), + )? + .unwrap_or_default(); let yaml_str = std::fs::read_to_string(&args.input_config)?; let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs index eda78f66..b6521b70 100644 --- a/asap-planner-rs/src/bin/optimizer_cli.rs +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -7,7 +7,7 @@ use std::path::PathBuf; use asap_planner::optimizer::{ - load_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset, + load_optional_selected_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset, }; use asap_planner::ControllerConfig; use clap::Parser; @@ -35,14 +35,21 @@ struct Args { #[arg(long = "rho", default_value = "1.0", value_parser = parse_positive_finite)] rho: f64, - /// Path to the atomic-cost table sketch-bench's `atomic-costs` subcommand - /// exports (see ASAPQuery#524, sketch-bench#30). Omitted: every + /// Path to the versioned atomic-cost document sketch-bench's `atomic-costs` + /// subcommand exports. Requires --atomic-cost-workload to select exactly + /// one measured workload profile. Omitted: every /// benchmarked-family candidate (CMS/HLL/KLL) is dropped, since there is /// no data to cost it at — only trivial accumulators and EXACT remain /// selectable. #[arg(long = "atomic-costs")] atomic_costs: Option, + /// JSON `profiles[].workload` value copied from the sketch-bench atomic-cost + /// document. This makes the empirical workload profile explicit and avoids + /// mixing costs from different traces or time windows. + #[arg(long = "atomic-cost-workload", requires = "atomic_costs")] + atomic_cost_workload: Option, + #[arg(short, long, action = clap::ArgAction::Count)] verbose: u8, } @@ -70,8 +77,11 @@ fn main() -> anyhow::Result<()> { let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; let dataset = SeriesDataset::from_path(&args.dataset)?; - let atomic_cost_table = match &args.atomic_costs { - Some(path) => load_atomic_cost_table(path)?, + let atomic_cost_table = match load_optional_selected_atomic_cost_table( + args.atomic_costs.as_deref(), + args.atomic_cost_workload.as_deref(), + )? { + Some(table) => table, None => { tracing::warn!( "no --atomic-costs supplied; CMS/HLL/KLL candidates will never be selected" diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index d5bbf547..7dc0bdd3 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -8,7 +8,7 @@ //! hand; `atomic_cost_entry_deserializes_sketch_benchs_documented_shape` //! below is a canary for drift. -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::path::Path; use promql_utilities::query_logics::enums::AggregationType; @@ -22,8 +22,58 @@ use super::constants::{ use super::cost_model::AtomicCosts; const CMS_HEAP_BENCHMARK: &str = "cms-heap-topk-regularpath-vector2d"; +pub const ATOMIC_COST_SCHEMA_VERSION: u32 = 1; + +/// Versioned atomic-cost document emitted by `approxbench atomic-costs`. +/// +/// A profile is deliberately selected before candidate resolution: costs from +/// different input workloads must never be mixed by a flat lookup. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AtomicCostDocument { + pub schema_version: u32, + pub profiles: Vec, +} #[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AtomicCostProfile { + pub workload: WorkloadDescription, + pub entries: Vec, +} + +/// The provenance of the input data on which atomic costs were measured. +/// Synthetic descriptions remain opaque because their generator schema evolves +/// independently; external profiles are represented explicitly for selection. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum WorkloadDescription { + Synthetic { description: Value }, + External(ExternalWorkload), +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExternalWorkload { + pub source: String, + pub dataset: String, + pub mode: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub key_columns: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub group_columns: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub variate: Option, + pub value_column: String, + pub window_start_ns: i64, + pub window_end_ns: i64, + pub records_loaded: u64, + pub source_timestamp_unit: String, + pub timestamp_unit: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct AtomicCostEntry { pub sketch: String, pub sketch_config: Value, @@ -31,16 +81,99 @@ pub struct AtomicCostEntry { pub insert_cpu_secs: f64, pub merge_cpu_secs: f64, pub query_cpu_secs: f64, + pub query_accuracy: BTreeMap, } pub type AtomicCostTable = Vec; -/// Read an `AtomicCostTable` exported by `sketch-bench atomic-costs`. -pub fn load_atomic_cost_table(path: &Path) -> anyhow::Result { +/// Parse a standalone JSON workload selector. The selector is the exact +/// `profiles[].workload` value copied from the benchmark artifact, making the +/// selected empirical input explicit in an offline planning run. +pub fn load_workload_selector(path: &Path) -> anyhow::Result { + let raw = std::fs::read_to_string(path).map_err(|e| { + anyhow::anyhow!( + "reading atomic-cost workload selector {}: {e}", + path.display() + ) + })?; + serde_json::from_str(&raw).map_err(|e| { + anyhow::anyhow!( + "parsing atomic-cost workload selector {}: {e}", + path.display() + ) + }) +} + +/// Read a versioned `sketch-bench atomic-costs` document and return entries +/// from exactly one requested workload profile. +pub fn load_atomic_cost_table( + path: &Path, + workload: &WorkloadDescription, +) -> anyhow::Result { let raw = std::fs::read_to_string(path) .map_err(|e| anyhow::anyhow!("reading atomic-cost table {}: {e}", path.display()))?; - serde_json::from_str(&raw) - .map_err(|e| anyhow::anyhow!("parsing atomic-cost table {}: {e}", path.display())) + let document: AtomicCostDocument = serde_json::from_str(&raw) + .map_err(|e| anyhow::anyhow!("parsing atomic-cost document {}: {e}", path.display()))?; + + if document.schema_version != ATOMIC_COST_SCHEMA_VERSION { + anyhow::bail!( + "unsupported atomic-cost schema_version {} in {} (this planner supports {})", + document.schema_version, + path.display(), + ATOMIC_COST_SCHEMA_VERSION + ); + } + + let matches: Vec<_> = document + .profiles + .iter() + .filter(|profile| profile.workload == *workload) + .collect(); + match matches.as_slice() { + [profile] => Ok(profile.entries.clone()), + [] => anyhow::bail!( + "no atomic-cost profile in {} matches workload selector {}", + path.display(), + serde_json::to_string(workload).unwrap_or_else(|_| "".into()) + ), + _ => anyhow::bail!( + "{} atomic-cost profiles in {} match workload selector {}; expected exactly one", + matches.len(), + path.display(), + serde_json::to_string(workload).unwrap_or_else(|_| "".into()) + ), + } +} + +/// Load the selector artifact and return the corresponding empirical table. +/// Offline callers use this single interface so selector validation cannot +/// drift between planner tools. +pub fn load_selected_atomic_cost_table( + document_path: &Path, + selector_path: &Path, +) -> anyhow::Result { + let workload = load_workload_selector(selector_path)?; + load_atomic_cost_table(document_path, &workload) +} + +/// Resolve the optional atomic-cost CLI inputs as one unit. A document without +/// its workload selector is invalid; callers can keep their no-document +/// fallback without duplicating that validation. +pub fn load_optional_selected_atomic_cost_table( + document_path: Option<&Path>, + selector_path: Option<&Path>, +) -> anyhow::Result> { + match (document_path, selector_path) { + (Some(document_path), Some(selector_path)) => { + load_selected_atomic_cost_table(document_path, selector_path).map(Some) + } + (Some(_), None) => anyhow::bail!( + "--atomic-cost-workload is required with --atomic-costs; \ + it must contain the selected profiles[].workload JSON value" + ), + (None, Some(_)) => anyhow::bail!("--atomic-cost-workload requires --atomic-costs"), + (None, None) => Ok(None), + } } /// sketch-bench's (algorithm, params) key for one of ASAPQuery's benchmarked @@ -306,6 +439,120 @@ fn valid_cost_entry(entry: &AtomicCostEntry) -> bool { mod tests { use super::*; + #[test] + fn loader_selects_only_the_requested_external_profile() { + let requested = WorkloadDescription::External(ExternalWorkload { + source: "google".into(), + dataset: "google/task_usage.csv.gz".into(), + mode: "grouped".into(), + key_columns: vec![], + group_columns: vec!["machine_id".into()], + variate: None, + value_column: "cpu_rate".into(), + window_start_ns: 10, + window_end_ns: 20, + records_loaded: 100, + source_timestamp_unit: "microseconds".into(), + timestamp_unit: "nanoseconds".into(), + }); + let other = WorkloadDescription::External(ExternalWorkload { + window_end_ns: 30, + ..match requested.clone() { + WorkloadDescription::External(workload) => workload, + WorkloadDescription::Synthetic { .. } => unreachable!(), + } + }); + let document = serde_json::json!({ + "schema_version": 1, + "profiles": [ + {"workload": other, "entries": []}, + {"workload": requested, "entries": [{ + "sketch": "kll-percall", + "sketch_config": {"algorithm": "kll-percall", "params": {"k": 200}}, + "mem_bytes_per_instance": 6400.0, + "insert_cpu_secs": 1e-8, + "merge_cpu_secs": 1e-3, + "query_cpu_secs": 1e-4, + "query_accuracy": {"mean_rank_err": 0.01} + }]} + ] + }); + let file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(file.path(), document.to_string()).unwrap(); + + let table = load_atomic_cost_table(file.path(), &requested).unwrap(); + + assert_eq!(table.len(), 1); + assert_eq!(table[0].sketch, "kll-percall"); + } + + #[test] + fn loader_rejects_an_ambiguous_or_incompatible_document() { + let workload = WorkloadDescription::Synthetic { + description: serde_json::json!({"name": "one"}), + }; + let file = tempfile::NamedTempFile::new().unwrap(); + + std::fs::write( + file.path(), + serde_json::json!({ + "schema_version": 2, + "profiles": [] + }) + .to_string(), + ) + .unwrap(); + let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + assert!(err.to_string().contains("schema_version 2")); + assert!(err.to_string().contains("supports 1")); + + std::fs::write( + file.path(), + serde_json::json!({ + "schema_version": 1, + "profiles": [ + {"workload": workload, "entries": []}, + {"workload": workload, "entries": []} + ] + }) + .to_string(), + ) + .unwrap(); + let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + assert!(err.to_string().contains("2 atomic-cost profiles")); + + std::fs::write( + file.path(), + serde_json::json!({ + "schema_version": 1, + "profiles": [{ + "workload": {"synthetic": {"description": {"name": "other"}}}, + "entries": [] + }] + }) + .to_string(), + ) + .unwrap(); + let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + assert!(err.to_string().contains("no atomic-cost profile")); + } + + #[test] + fn atomic_cost_entry_requires_query_accuracy() { + let json = r#"{"sketch":"kll-percall","sketch_config":null,"mem_bytes_per_instance":1.0,"insert_cpu_secs":1.0,"merge_cpu_secs":1.0,"query_cpu_secs":1.0}"#; + assert!(serde_json::from_str::(json).is_err()); + } + + #[test] + fn optional_loader_rejects_an_unselected_document() { + let document = tempfile::NamedTempFile::new().unwrap(); + let err = + load_optional_selected_atomic_cost_table(Some(document.path()), None).unwrap_err(); + assert!(err + .to_string() + .contains("--atomic-cost-workload is required")); + } + fn cms_entry(depth: i64, width: i64) -> AtomicCostEntry { AtomicCostEntry { sketch: "cms-fastpath-vector2d".into(), @@ -317,6 +564,7 @@ mod tests { insert_cpu_secs: 8e-9, merge_cpu_secs: 4.5e-4, query_cpu_secs: 7.8e-8, + query_accuracy: BTreeMap::new(), } } @@ -338,6 +586,7 @@ mod tests { insert_cpu_secs: 2.0, merge_cpu_secs: 4.0, query_cpu_secs: 8.0, + query_accuracy: BTreeMap::new(), } } @@ -351,10 +600,8 @@ mod tests { #[test] fn atomic_cost_entry_deserializes_sketch_benchs_documented_shape() { - // Pinned against a real row sketch-bench's `atomic-costs` subcommand - // actually emitted (out/atomic_costs.json, cms-fastpath-vector2d - // rows=3 cols=1024) -- a canary for the two structs drifting apart. - let json = r#"{"sketch":"cms-fastpath-vector2d","sketch_config":{"algorithm":"cms-fastpath-vector2d","params":{"cols":1024,"rows":3}},"mem_bytes_per_instance":12288.0,"insert_cpu_secs":8.484689139741214e-9,"merge_cpu_secs":0.00045364040539336466,"query_cpu_secs":7.799774697708031e-8}"#; + // Pinned against the current sketch-bench atomic-cost entry shape. + let json = r#"{"sketch":"cms-fastpath-vector2d","sketch_config":{"algorithm":"cms-fastpath-vector2d","params":{"cols":1024,"rows":3}},"mem_bytes_per_instance":12288.0,"insert_cpu_secs":8.484689139741214e-9,"merge_cpu_secs":0.00045364040539336466,"query_cpu_secs":7.799774697708031e-8,"query_accuracy":{"relative_error":0.01}}"#; let entry: AtomicCostEntry = serde_json::from_str(json).expect("documented shape parses"); assert_eq!(entry.sketch, "cms-fastpath-vector2d"); assert_eq!(entry.mem_bytes_per_instance, 12288.0); @@ -487,6 +734,7 @@ mod tests { insert_cpu_secs: 1.68e-9, merge_cpu_secs: 2.76e-4, query_cpu_secs: 1.23e-4, + query_accuracy: BTreeMap::new(), }]; let hll_params = HashMap::from([("precision".to_string(), Value::from(14u64))]); assert!(resolve_atomic_costs(&hll_table, AggregationType::HLL, &hll_params).is_some()); @@ -498,6 +746,7 @@ mod tests { insert_cpu_secs: 1.6e-8, merge_cpu_secs: 1.0e-3, query_cpu_secs: 1.6e-4, + query_accuracy: BTreeMap::new(), }]; let kll_params = HashMap::from([("K".to_string(), Value::from(200u64))]); assert!( diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs index ecbdd2c5..a533a559 100644 --- a/asap-planner-rs/src/optimizer/greedy.rs +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -225,6 +225,7 @@ mod tests { insert_cpu_secs: 0.0, merge_cpu_secs: 0.0, query_cpu_secs: 0.0, + query_accuracy: std::collections::BTreeMap::new(), }]; let aqe = make_aqe(Statistic::Topk, 60_000, 60_000, 1.0 / 60.0); let solution = greedy_assign( diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs index e88fc759..521db371 100644 --- a/asap-planner-rs/src/optimizer/mod.rs +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -12,7 +12,9 @@ pub mod translator; pub use aqe_extractor::{extract_aqes, RQE}; pub use atomic_costs::{ - load_atomic_cost_table, resolve_atomic_costs, AtomicCostEntry, AtomicCostTable, + load_atomic_cost_table, load_optional_selected_atomic_cost_table, + load_selected_atomic_cost_table, resolve_atomic_costs, AtomicCostEntry, AtomicCostTable, + ExternalWorkload, WorkloadDescription, }; pub use candidate_gen::{ enumerate_candidates, enumerate_candidates_with_label_group_count, CandidateConfig, diff --git a/asap-query-engine/src/precompute_engine/engine.rs b/asap-query-engine/src/precompute_engine/engine.rs index 1eb6e532..4fd67cc7 100644 --- a/asap-query-engine/src/precompute_engine/engine.rs +++ b/asap-query-engine/src/precompute_engine/engine.rs @@ -1,7 +1,7 @@ use crate::data_model::{AggregationType, StreamingConfig}; use crate::precompute_engine::accumulator_factory::create_accumulator_updater; use crate::precompute_engine::config::PrecomputeEngineConfig; -use crate::precompute_engine::ingest_source::{IngestContext, IngestSource}; +use crate::precompute_engine::ingest_source::{IngestContext, IngestSource, RoutingConfigSet}; use crate::precompute_engine::output_sink::OutputSink; use crate::precompute_engine::series_router::{SeriesRouter, WorkerMessage}; use crate::precompute_engine::worker::{Worker, WorkerRuntimeConfig}; @@ -22,31 +22,36 @@ pub struct PrecomputeWorkerDiagnostics { /// A cloneable handle for applying runtime config updates to a running engine. /// /// Obtained via `PrecomputeEngine::handle()` before calling `run()`. -/// Calling `update_streaming_config` swaps the ingest handler's agg_configs +/// Calling `update_streaming_config` swaps the ingest handler's routing configs /// atomically (lock-free via ArcSwap) and broadcasts the new map to all workers. pub struct PrecomputeEngineHandle { router: SeriesRouter, - ingest_agg_configs: Arc>>>, + ingest_routing_configs: Arc>, } impl PrecomputeEngineHandle { /// Apply a new streaming config to the running engine. /// - /// Updates the ingest handler's agg_configs via a lock-free ArcSwap store, + /// Updates the ingest handler's routing configs via a lock-free ArcSwap store, /// then broadcasts the new config map to all workers via their message channels. pub async fn update_streaming_config( &self, config: &StreamingConfig, ) -> Result<(), Box> { - let agg_configs_map: HashMap> = config - .get_all_aggregation_configs() + let routing_configs = RoutingConfigSet::from_streaming_config(config) + .map_err(|error| -> Box { error.into() })?; + let agg_configs_map: HashMap> = routing_configs + .configs() .iter() - .map(|(&id, cfg)| (id, Arc::new(cfg.clone()))) + .map(|routing_config| { + ( + routing_config.config.aggregation_id, + routing_config.config.clone(), + ) + }) .collect(); - let agg_configs_vec: Vec> = - agg_configs_map.values().cloned().collect(); - self.ingest_agg_configs.store(Arc::new(agg_configs_vec)); + self.ingest_routing_configs.store(Arc::new(routing_configs)); self.router .broadcast_update_agg_configs(agg_configs_map) .await?; @@ -73,8 +78,8 @@ pub struct PrecomputeEngine { /// Channels created at construction so handle() can be extracted before run(). senders: Vec>, receivers: Option>>, - /// Shared ingest agg_configs, swappable at runtime. - ingest_agg_configs: Arc>>>, + /// Shared ingest routing configs, swappable at runtime. + ingest_routing_configs: Arc>, /// Test-support wall-clock override, applied to every spawned worker. /// See `with_now_ms_fn`. `None` in production — each worker keeps its /// default `SystemTime::now`-backed clock. @@ -108,12 +113,7 @@ impl PrecomputeEngine { receivers.push(rx); } - let agg_configs_vec: Vec> = streaming_config - .get_all_aggregation_configs() - .values() - .map(|cfg| Arc::new(cfg.clone())) - .collect(); - let ingest_agg_configs = Arc::new(ArcSwap::from_pointee(agg_configs_vec)); + let ingest_routing_configs = Arc::new(ArcSwap::from_pointee(RoutingConfigSet::empty())); Self { config, @@ -123,7 +123,7 @@ impl PrecomputeEngine { sources, senders, receivers: Some(receivers), - ingest_agg_configs, + ingest_routing_configs, now_ms_fn: None, } } @@ -148,7 +148,7 @@ impl PrecomputeEngine { pub fn handle(&self) -> PrecomputeEngineHandle { PrecomputeEngineHandle { router: SeriesRouter::new(self.senders.clone()), - ingest_agg_configs: self.ingest_agg_configs.clone(), + ingest_routing_configs: self.ingest_routing_configs.clone(), } } @@ -156,6 +156,9 @@ impl PrecomputeEngine { /// ingest sources, then blocks until shutdown. pub async fn run(mut self) -> Result<(), Box> { validate_startup_aggregation_configs(&self.streaming_config)?; + let routing_configs = RoutingConfigSet::from_streaming_config(&self.streaming_config) + .map_err(|error| -> Box { error.into() })?; + self.ingest_routing_configs.store(Arc::new(routing_configs)); let num_workers = self.config.num_workers; @@ -212,7 +215,7 @@ impl PrecomputeEngine { // handle.update_streaming_config() is immediately visible to every source. let ctx = IngestContext { router: router.clone(), - agg_configs: self.ingest_agg_configs.clone(), + routing_configs: self.ingest_routing_configs.clone(), pass_raw_samples: self.config.pass_raw_samples, }; @@ -319,6 +322,41 @@ mod tests { } } + #[tokio::test] + async fn run_rejects_invalid_spatial_filter_before_starting_workers() { + let config = AggregationConfig::new( + 1, + AggregationType::Sum, + "Sum".to_string(), + HashMap::new(), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + String::new(), + 1_000, + 1_000, + WindowType::Tumbling, + r#"{job=}"#.to_string(), + "requests_total".to_string(), + None, + None, + None, + None, + ); + let engine = PrecomputeEngine::new( + PrecomputeEngineConfig::default(), + Arc::new(StreamingConfig::new(HashMap::from([(1, config)]))), + Arc::new(NoopOutputSink::new()), + vec![Box::new(ShutdownSource)], + ); + + let error = engine + .run() + .await + .expect_err("invalid spatial filter must reject startup"); + assert!(error.to_string().contains("invalid spatialFilter")); + } + #[tokio::test] async fn run_rejects_invalid_cms_subtype_before_starting_workers() { let mut parameters = HashMap::new(); @@ -338,7 +376,7 @@ mod tests { 1_000, 1_000, WindowType::Tumbling, - "requests_total".to_string(), + String::new(), "requests_total".to_string(), None, None, @@ -381,7 +419,7 @@ mod tests { 1_000, 1_000, WindowType::Tumbling, - "requests_total".to_string(), + String::new(), "requests_total".to_string(), None, None, @@ -428,7 +466,7 @@ mod tests { 1_000, 1_000, WindowType::Tumbling, - "requests_total".to_string(), + String::new(), "requests_total".to_string(), None, None, @@ -472,7 +510,7 @@ mod tests { 1_000, 1_000, WindowType::Tumbling, - "requests_total".to_string(), + String::new(), "requests_total".to_string(), None, None, diff --git a/asap-query-engine/src/precompute_engine/ingest_source.rs b/asap-query-engine/src/precompute_engine/ingest_source.rs index 9fb46dc2..3b4670fa 100644 --- a/asap-query-engine/src/precompute_engine/ingest_source.rs +++ b/asap-query-engine/src/precompute_engine/ingest_source.rs @@ -3,6 +3,9 @@ use crate::precompute_engine::series_router::{SeriesRouter, WorkerMessage}; use crate::precompute_engine::worker::{extract_metric_name, parse_labels_from_series_key}; use arc_swap::ArcSwap; use asap_types::aggregation_config::AggregationConfig; +use asap_types::streaming_config::StreamingConfig; +use promql_parser::label::Matcher; +use promql_parser::parser::{parse, Expr}; use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -13,15 +16,116 @@ use tracing::{debug, warn}; #[derive(Clone)] pub struct IngestContext { pub(crate) router: SeriesRouter, - /// Aggregation configs for group-key extraction. - /// Wrapped in Arc so the same ArcSwap is shared with PrecomputeEngineHandle. - /// The handle calls ArcSwap::store() to push a new Vec; this context sees it - /// immediately via the shared Arc pointer (lock-free on the read path). - pub(crate) agg_configs: Arc>>>, + /// Aggregation configs and their compiled spatial filters. + /// Shared with `PrecomputeEngineHandle`, whose swaps are immediately visible + /// on this lock-free read path. + pub(crate) routing_configs: Arc>, /// When true, skip group-key extraction and pass raw samples through. pub(crate) pass_raw_samples: bool, } +/// Aggregation configuration enriched with its precompiled spatial filter. +#[derive(Clone)] +pub(crate) struct RoutingAggregationConfig { + pub(crate) config: Arc, + spatial_matchers: Vec, +} + +impl RoutingAggregationConfig { + fn matches_spatial_filter(&self, labels: &HashMap<&str, &str>) -> bool { + self.spatial_matchers.iter().all(|matcher| { + let value = labels.get(matcher.name.as_str()).copied().unwrap_or(""); + matcher.is_match(value) + }) + } +} + +/// The complete, atomically swappable routing configuration for ingest. +pub(crate) struct RoutingConfigSet { + configs: Vec, +} + +impl RoutingConfigSet { + pub(crate) fn empty() -> Self { + Self { + configs: Vec::new(), + } + } + + pub(crate) fn from_streaming_config(config: &StreamingConfig) -> Result { + Self::from_aggregation_configs(config.get_all_aggregation_configs().values().cloned()) + } + + pub(crate) fn from_aggregation_configs( + configs: impl IntoIterator, + ) -> Result { + let configs = configs + .into_iter() + .map(|config| { + let spatial_matchers = compile_spatial_filter(&config)?; + Ok(RoutingAggregationConfig { + config: Arc::new(config), + spatial_matchers, + }) + }) + .collect::, String>>()?; + Ok(Self { configs }) + } + + pub(crate) fn configs(&self) -> &[RoutingAggregationConfig] { + &self.configs + } +} + +fn compile_spatial_filter(config: &AggregationConfig) -> Result, String> { + // SQL aggregation configs do not use PromQL label selectors. + if config.table_name.is_some() || config.spatial_filter.trim().is_empty() { + return Ok(Vec::new()); + } + + let filter = config.spatial_filter.trim(); + let selector_body = if filter.starts_with('{') || filter.ends_with('}') { + if !filter.starts_with('{') || !filter.ends_with('}') { + return Err(format!( + "aggregation_id {} has invalid spatialFilter {:?}: unmatched selector braces", + config.aggregation_id, config.spatial_filter + )); + } + filter.to_string() + } else { + format!("{{{filter}}}") + }; + let selector = format!("{}{}", config.metric, selector_body); + let Expr::VectorSelector(vector_selector) = parse(&selector).map_err(|error| { + format!( + "aggregation_id {} has invalid spatialFilter {:?}: {error}", + config.aggregation_id, config.spatial_filter + ) + })? + else { + return Err(format!( + "aggregation_id {} has invalid spatialFilter {:?}: expected a vector selector", + config.aggregation_id, config.spatial_filter + )); + }; + + let matchers = vector_selector.matchers; + if !matchers.or_matchers.is_empty() { + return Err(format!( + "aggregation_id {} spatialFilter must not use selector-level or", + config.aggregation_id + )); + } + let matchers = matchers.matchers; + if matchers.iter().any(|matcher| matcher.name == "__name__") { + return Err(format!( + "aggregation_id {} spatialFilter must not match __name__; use metric instead", + config.aggregation_id + )); + } + Ok(matchers) +} + /// An ingest source for the precompute engine. /// /// Implementors decode incoming data (HTTP, file, etc.) and push it @@ -34,8 +138,10 @@ pub trait IngestSource: Send + Sync { ) -> Result<(), Box>; } -pub(crate) fn extract_group_key(series_key: &str, config: &AggregationConfig) -> String { - let labels = parse_labels_from_series_key(series_key); +pub(crate) fn extract_group_key( + labels: &HashMap<&str, &str>, + config: &AggregationConfig, +) -> String { let mut values = Vec::new(); for label_name in &config.grouping_labels.labels { if let Some(val) = labels.get(label_name.as_str()) { @@ -89,8 +195,8 @@ pub(crate) async fn route_decoded_samples( type SampleTuple = (String, i64, f64); let mut by_group: HashMap> = HashMap::new(); - // Load agg_configs once per request (lock-free ArcSwap read). - let agg_configs = ctx.agg_configs.load(); + // Load routing configs once per request (lock-free ArcSwap read). + let routing_configs = ctx.routing_configs.load(); // On first batch: log config metrics vs sample metric to diagnose mismatches. static FIRST_BATCH_LOGGED: AtomicBool = AtomicBool::new(false); @@ -100,10 +206,11 @@ pub(crate) async fn route_decoded_samples( warn!( sample_metric, sample_labels = %first.labels, - num_agg_configs = agg_configs.len(), + num_agg_configs = routing_configs.configs().len(), "routing: first batch diagnostic" ); - for cfg in agg_configs.iter() { + for routing_config in routing_configs.configs() { + let cfg = &routing_config.config; warn!( agg_id = cfg.aggregation_id, config_metric = %cfg.metric, @@ -118,16 +225,16 @@ pub(crate) async fn route_decoded_samples( let mut matched_samples: usize = 0; for s in &samples { let metric_name = extract_metric_name(&s.labels); - for config in agg_configs.iter() { - if config.metric != metric_name - && config.spatial_filter_normalized != metric_name - && config.spatial_filter != metric_name - && config.table_name.as_deref() != Some(metric_name) + let labels = parse_labels_from_series_key(&s.labels); + for routing_config in routing_configs.configs() { + let config = &routing_config.config; + if (config.metric != metric_name && config.table_name.as_deref() != Some(metric_name)) + || !routing_config.matches_spatial_filter(&labels) { continue; } matched_samples += 1; - let group_key = extract_group_key(&s.labels, config); + let group_key = extract_group_key(&labels, config); by_group .entry((config.aggregation_id, group_key)) .or_default() @@ -159,3 +266,256 @@ pub(crate) async fn route_decoded_samples( .await?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::enums::{AggregationType, WindowType}; + use promql_utilities::data_model::key_by_label_names::KeyByLabelNames; + use tokio::sync::mpsc::error::TryRecvError; + + fn aggregation_config(spatial_filter: &str) -> AggregationConfig { + AggregationConfig::new( + 7, + AggregationType::Sum, + "Sum".to_string(), + HashMap::new(), + KeyByLabelNames::new(vec![]), + KeyByLabelNames::new(vec![]), + KeyByLabelNames::new(vec![]), + String::new(), + 1_000, + 1_000, + WindowType::Tumbling, + spatial_filter.to_string(), + "cpu_usage".to_string(), + None, + None, + None, + None, + ) + } + + fn sample(labels: &str, timestamp_ms: i64, value: f64) -> DecodedSample { + DecodedSample { + labels: labels.to_string(), + timestamp_ms, + value, + } + } + + async fn routed_samples( + spatial_filter: &str, + input_samples: Vec, + ) -> Vec<(String, i64, f64)> { + let (sender, mut receiver) = tokio::sync::mpsc::channel(2); + let context = IngestContext { + router: SeriesRouter::new(vec![sender]), + routing_configs: Arc::new(ArcSwap::from_pointee( + RoutingConfigSet::from_aggregation_configs(vec![aggregation_config( + spatial_filter, + )]) + .expect("valid test spatial filter"), + )), + pass_raw_samples: false, + }; + + route_decoded_samples(&context, input_samples, Instant::now()) + .await + .expect("routing should succeed"); + + let message = receiver.recv().await.expect("one routed message"); + let WorkerMessage::GroupSamples { + agg_id, + group_key, + samples, + .. + } = message + else { + panic!("normal routing should produce GroupSamples"); + }; + + assert_eq!(agg_id, 7); + assert_eq!(group_key, ""); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); + samples + } + + #[tokio::test] + async fn matching_metric_with_no_spatial_filter_routes_once_to_its_aggregation() { + let samples = routed_samples( + "", + vec![sample("cpu_usage{instance=\"a\",job=\"api\"}", 1_000, 42.0)], + ) + .await; + assert_eq!( + samples, + vec![( + ("cpu_usage{instance=\"a\",job=\"api\"}").to_string(), + 1_000, + 42.0 + )] + ); + } + + #[tokio::test] + async fn planner_style_spatial_filter_routes_only_equal_label_values() { + let samples = routed_samples( + r#"job="api""#, + vec![ + sample("cpu_usage{job=\"api\"}", 1_000, 1.0), + sample("cpu_usage{job=\"worker\"}", 2_000, 2.0), + ], + ) + .await; + + assert_eq!( + samples, + vec![("cpu_usage{job=\"api\"}".to_string(), 1_000, 1.0)] + ); + } + + #[tokio::test] + async fn inequality_spatial_filter_routes_only_different_label_values() { + let samples = routed_samples( + r#"{job!="api"}"#, + vec![ + sample("cpu_usage{job=\"api\"}", 1_000, 1.0), + sample("cpu_usage{job=\"worker\"}", 2_000, 2.0), + sample("cpu_usage{instance=\"a\"}", 3_000, 3.0), + ], + ) + .await; + + assert_eq!( + samples, + vec![ + ("cpu_usage{job=\"worker\"}".to_string(), 2_000, 2.0), + ("cpu_usage{instance=\"a\"}".to_string(), 3_000, 3.0), + ] + ); + } + + #[tokio::test] + async fn non_empty_regex_spatial_filter_drops_empty_label_values() { + let samples = routed_samples( + r#"{job=~".+"}"#, + vec![ + sample("cpu_usage{job=\"api\"}", 1_000, 1.0), + sample("cpu_usage{job=\"\"}", 2_000, 2.0), + sample("cpu_usage{instance=\"a\"}", 3_000, 3.0), + ], + ) + .await; + + assert_eq!( + samples, + vec![("cpu_usage{job=\"api\"}".to_string(), 1_000, 1.0)] + ); + } + + #[tokio::test] + async fn alternation_regex_spatial_filter_routes_only_list_members() { + let samples = routed_samples( + r#"{job=~"user-service|order-service|payment-service"}"#, + vec![ + sample("cpu_usage{job=\"user-service\"}", 1_000, 1.0), + sample("cpu_usage{job=\"inventory-service\"}", 2_000, 2.0), + sample("cpu_usage{job=\"payment-service\"}", 3_000, 3.0), + ], + ) + .await; + + assert_eq!( + samples, + vec![ + ("cpu_usage{job=\"user-service\"}".to_string(), 1_000, 1.0), + ("cpu_usage{job=\"payment-service\"}".to_string(), 3_000, 3.0,), + ] + ); + } + + #[tokio::test] + async fn starts_with_regex_spatial_filter_routes_only_matching_prefixes() { + let samples = routed_samples( + r#"{interface=~"eth0.*"}"#, + vec![ + sample("cpu_usage{interface=\"eth0\"}", 1_000, 1.0), + sample("cpu_usage{interface=\"eth0.100\"}", 2_000, 2.0), + sample("cpu_usage{interface=\"ens3\"}", 3_000, 3.0), + ], + ) + .await; + + assert_eq!( + samples, + vec![ + ("cpu_usage{interface=\"eth0\"}".to_string(), 1_000, 1.0), + ("cpu_usage{interface=\"eth0.100\"}".to_string(), 2_000, 2.0,), + ] + ); + } + + #[tokio::test] + async fn sample_routes_to_every_matching_spatial_filter_config() { + let first = aggregation_config(r#"{job="api"}"#); + let mut second = aggregation_config(r#"{status=~"5.."}"#); + second.aggregation_id = 8; + let (sender, mut receiver) = tokio::sync::mpsc::channel(2); + let context = IngestContext { + router: SeriesRouter::new(vec![sender]), + routing_configs: Arc::new(ArcSwap::from_pointee( + RoutingConfigSet::from_aggregation_configs(vec![first, second]) + .expect("valid test spatial filters"), + )), + pass_raw_samples: false, + }; + + route_decoded_samples( + &context, + vec![sample("cpu_usage{job=\"api\",status=\"500\"}", 1_000, 1.0)], + Instant::now(), + ) + .await + .expect("routing should succeed"); + + let mut aggregation_ids = Vec::new(); + for _ in 0..2 { + let WorkerMessage::GroupSamples { + agg_id, samples, .. + } = receiver.recv().await.expect("one routed message") + else { + panic!("normal routing should produce GroupSamples"); + }; + assert_eq!( + samples, + vec![( + "cpu_usage{job=\"api\",status=\"500\"}".to_string(), + 1_000, + 1.0, + )] + ); + aggregation_ids.push(agg_id); + } + aggregation_ids.sort_unstable(); + assert_eq!(aggregation_ids, vec![7, 8]); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); + } + + #[test] + fn invalid_spatial_filters_are_rejected_when_routing_configs_are_built() { + let invalid_syntax = + RoutingConfigSet::from_aggregation_configs(vec![aggregation_config(r#"{job=}"#)]); + assert!(invalid_syntax.is_err()); + + let metric_matcher = RoutingConfigSet::from_aggregation_configs(vec![aggregation_config( + r#"{__name__="other_metric"}"#, + )]); + assert!(metric_matcher.is_err()); + + let selector_or = RoutingConfigSet::from_aggregation_configs(vec![aggregation_config( + r#"{job="api" or job="worker"}"#, + )]); + assert!(selector_or.is_err()); + } +} diff --git a/asap-query-engine/tests/e2e_netflow_single_second.rs b/asap-query-engine/tests/e2e_netflow_single_second.rs index 3196f2c6..54628d73 100644 --- a/asap-query-engine/tests/e2e_netflow_single_second.rs +++ b/asap-query-engine/tests/e2e_netflow_single_second.rs @@ -40,7 +40,7 @@ fn netflow_agg_config(metric: &str, window_size_ms: u64) -> AggregationConfig { window_size_ms, 0, WindowType::Tumbling, - metric.to_string(), + String::new(), metric.to_string(), None, None, diff --git a/asap-query-engine/tests/e2e_precompute_equivalence.rs b/asap-query-engine/tests/e2e_precompute_equivalence.rs index db765476..b7ce962d 100644 --- a/asap-query-engine/tests/e2e_precompute_equivalence.rs +++ b/asap-query-engine/tests/e2e_precompute_equivalence.rs @@ -80,7 +80,7 @@ fn make_agg_config_full( window_size_ms, slide_interval_ms, window_type, - metric.to_string(), + String::new(), metric.to_string(), None, None, diff --git a/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs b/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs index 150f3dc9..72dfb104 100644 --- a/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs +++ b/asap-query-engine/tests/e2e_precompute_wall_clock_fallback_active_ingest.rs @@ -60,7 +60,7 @@ fn netflow_agg_config(metric: &str, window_size_ms: u64) -> AggregationConfig { window_size_ms, 0, WindowType::Tumbling, - metric.to_string(), + String::new(), metric.to_string(), None, None,