diff --git a/.design_docs/sketch-config-optimization-formulation.md b/.design_docs/sketch-config-optimization-formulation.md new file mode 100644 index 00000000..f101b64c --- /dev/null +++ b/.design_docs/sketch-config-optimization-formulation.md @@ -0,0 +1,493 @@ +# Sketch/Streaming Config Selection as an Optimization Problem + +## Motivation + +`asap-planner-rs` currently hardcodes the decisions that should be the output +of an optimization: + +- **Sketch choice** — `planner/sketch.rs`: a fixed `Statistic → AggregationType` + map (`map_statistic_to_precompute_operator`). +- **Sketch parameters** — `planner/sketch.rs:6-14`: fixed constants + (e.g. CMS depth=3/width=1024), overridable via `SketchParameterOverrides` + but not chosen algorithmically. +- **Window size** — `planner/window.rs`: always tumbling + (`should_use_sliding_window()` hardcoded to `false`, "sliding windows + crash Arroyo"), size = effective repeat interval or scrape interval. +- **Query support** — `planner/promql.rs:163-169`: `is_supported()` is a + boolean AST-pattern match, with no notion of cost/accuracy tradeoff. + +There is no mechanism for one streaming config to serve multiple queries +(`CAPABILITY_MATCHING_DESIGN.md` is exploring this separately), and no +connection between `sketch-bench`'s empirical cost/accuracy measurements and +the planner's decisions — `sketch-bench` is a standalone offline benchmark +today. + +This doc formalizes sketch/config/window selection as a single optimization +problem (v1 scope: static, one-shot — see "Out of scope for v1" below). + +## Inputs + +- **RQE** (repeating query expression): `r = (QE_r, T_r)` — a query + expression repeating every `T_r` seconds (e.g. a dashboard panel's refresh + interval). +- **QE** (query expression): a tree of AQEs joined by binary arithmetic + operators. E.g. `quantile_over_time(0.5, data[1m]) / quantile_over_time(0.9, data[5m])` + is one QE with two AQEs. +- **AQE** (atomic query expression): a single aggregation + (metric + label set + statistic + spatial op + range vector). AQEs are + deduplicated by structural identity across all RQEs. The same AQE can + appear in multiple RQEs, each with its own `T_r`. +- `range_a` — the lookback duration baked into AQE `a`'s range vector + selector (e.g. the `5m` in `data[5m]`). This is a static property of the + AQE itself, not something the optimizer estimates. +- `θ_a` — profiled cardinality/skew characteristics of `a`. Computed by an + external profiling/estimation step (sampling) **before** the optimization + runs; consumed as opaque input data. (How to build this profiling step is + punted — see below.) +- `ρ_g` — real-world arrival rate of data feeding streaming config `g` + (e.g. Prometheus scrape rate × number of active series matching `g`'s + metric/label set). Defined on `g`'s own metric/label set, not on any + particular assigned AQE, for the same reason retention is a field of `g` + rather than derived from the assignment. Needed because sketch-bench's + throughput numbers measure a sketch's *capacity* (max items/sec it can + absorb), not the cost at the actual production rate. `IngestCost(g)` + should be computed as `per_item_cost(g) × ρ_g` where `per_item_cost(g) = + total_CPU_time / item_count` from sketch-bench's benchmark run, not read + directly off the benchmark's reported throughput. +- `ε_a` — required accuracy tolerance for `a`, derived from the query's + semantics. +- sketch-bench data: empirical (and where available, analytic) cost/accuracy + numbers as a function of `(sketch_type, params, dataset characteristics)`. + +## Sets + +- `A = {a_1, ..., a_m}` — all AQEs (deduplicated). +- `R = {r_1, ..., r_n}` — all RQEs, each with `A_r ⊆ A`. +- `R_a = {r ∈ R : a ∈ A_r}` — RQEs that reference AQE `a`. +- `G = {g_1, ..., g_k}` — candidate streaming configs (see "Generating G"). + Each `g = (sketch_type, params, ingest_type, W, S, retention_windows, + metric, label_set, spatial_filter)` where `ingest_type ∈ {tumbling, + sliding}`, `W` = window size, `S` = slide interval (`S = W` for tumbling, + `S < W` for sliding). For every AQE `a`, `G` includes a dedicated + `EXACT_a` config: no streaming dataflow, answers `a` directly from + already-retained raw data at query time (`IngestCost = 0`, `Error = 0`). + +## Derived quantities + +- `f_a = Σ_{r ∈ R_a} 1/T_r` — query frequency for AQE `a` (queries/sec), + combining all RQEs that reference it. +- `IngestCost(g) = w₁·CPU_ingest(g) + w₂·Mem_ingest(g)` — a **rate** (cost + per second), purely a function of `g`. Crucially does **not** depend on + which AQEs are assigned to `g` (see "Why retention is part of g"). + Concrete formulas given in "Analytical Cost Model" below. +- `QueryCost(a,g) = w₃·CPU_query(a,g) + w₄·Mem_query(a,g)` — cost of **one** + query of `a` against `g`, a function of `(a, g, θ_a, range_a)`. Per-event + cost, not a rate. (Latency is *not* folded in here — see optional latency + constraint below.) Concrete formulas given in "Analytical Cost Model" below. +- `Error(a,g,θ_a)` — accuracy of `g` answering `a` given `a`'s real profile. + Use closed-form bounds where the sketch has one (e.g. CMS's ε-δ + guarantee); empirical sketch-bench lookup at the nearest profiled point + otherwise. `Error(a, EXACT_a, θ_a) = 0` always. +- `Feasible(a,g) ∈ {0,1}`: `g` can answer `a` if and only if — + 1. `g`'s label set/metric/spatial-filter capability covers what `a` needs + (capability matching — generalizes `CAPABILITY_MATCHING_DESIGN.md`). + Directional: `g` computed at a finer label granularity can serve an + `a` that wants a coarser rollup (merge at query time, costed in + `QueryCost`); the reverse is never feasible — once `g` aggregates + away a label, that information is gone and no `a` needing that finer + granularity can be served from it. + 2. `retention_windows(g) · window_size(g) ≥ range_a` (g retains enough + history to answer a's lookback range), + 3. `Error(a,g,θ_a) ≤ ε_a`, + 4. A valid (ingest type, query method) combination exists for `(g, a)` — + see the compatibility table in "Analytical Cost Model". Concretely: + sliding ingestion with `W < range_a` is always infeasible; tumbling + with `W < range_a` requires `mergeable(s)` or `subtractable(s)`. + + Folding all four checks into `Feasible` means none needs to appear as a + separate constraint later — they just determine which `(a,g)` pairs are + legal. + +## Analytical Cost Model + +This section gives concrete formulas for `IngestCost(g)` and `QueryCost(a,g)` +in terms of atomic per-sketch costs from sketch-bench and structural +multipliers derived from window configuration and sketch algebraic properties. + +### Sketch algebraic properties + +Three boolean properties per sketch type `s`. Exact values per sketch family +are to be catalogued separately (deferred — see "Out of scope for v1"): + +- **`mergeable(s)`** — two instances can be combined into one representing the + union of their input streams. Required for multi-window range queries via + merge at query time. +- **`subtractable(s)`** — one instance can be approximately subtracted from + another to give the difference of their input streams. Enables O(1)-cost + range queries via prefix-sum checkpoints regardless of `range_a/W`. Only + valid with tumbling ingestion: sliding ingestion's overlapping retained + windows are non-prefix-summable, ruling out subtraction. +- **`subpopulation_aware(s)`** — one sketch instance handles all label-group + keys internally (e.g. CMS maps arbitrary string keys to counts in a single + structure). When `false`, a separate instance is maintained per distinct + label-group combination and costs scale with `N_g`. + +### Atomic costs from sketch-bench + +For sketch type `s`, params `p`: + +| Symbol | Meaning | +|--------|---------| +| `mem(s,p)` | memory of one sketch instance | +| `insert_cpu(s,p)` | CPU per item inserted | +| `merge_cpu(s,p)` | CPU to merge two instances (`mergeable(s)` only) | +| `subtract_cpu(s,p)` | CPU to subtract two instances (`subtractable(s)` only) | +| `query_cpu(s,p)` | CPU to answer one query from one instance | + +These are extracted from sketch-bench as: `per_item_cost = total_CPU_time / +item_count` for insert; direct measurements for merge/subtract/query. See +"Data gap" section — cardinality sweeps needed before these are available +across the full parameter grid. + +### Label-group scaling multiplier + +``` +N(s,g) = 1 if subpopulation_aware(s) + N_g otherwise +``` + +where `N_g` = number of distinct label-group combinations for `g`'s +metric/label set (from `θ_a`). Applies uniformly to memory, query CPU, and +merge/subtract CPU — both subpopulation-aware and non-subpopulation-aware +sketches use the same formulas below with this multiplier. + +### Window constraints (on `g`'s parameters) + +``` +W ≤ range_a — a sketch must not over-cover the query range; data from + before range_a seconds ago is permanently absorbed and + cannot be extracted after the fact +W ≤ T_r — freshness: a completed window must be available for each + query repetition cycle +``` + +For sliding ingestion: "neither" (the only valid sliding query method) +requires `range_a ≤ W`, combined with `W ≤ range_a` this forces `W = +range_a`. Sliding ingestion is therefore only valid when `range_a ≤ T_r`. + +### Valid (ingest type, query method) combinations + +The query method is fully determined by ingest type, the relationship of `W` +to `range_a`, and the sketch's algebraic properties. It is not an independent +decision variable: + +| Ingest type | Condition | Query method | +|-------------|-----------|--------------| +| Tumbling | `W = range_a` | **Neither** — direct read from one window | +| Tumbling | `W < range_a`, `subtractable(s)` | **Subtract** | +| Tumbling | `W < range_a`, `mergeable(s)`, not subtractable | **Merge** | +| Tumbling | `W < range_a`, neither property | **Infeasible** | +| Sliding | `W = range_a` (forced) | **Neither** | +| Sliding | `W < range_a` | **Infeasible** — overlapping retained windows cannot be merged or subtracted | + +### Cost formulas + +Let `n = ⌈range_a / W⌉` (number of retained windows needed to cover `range_a`). + +#### Ingestion memory (steady-state, two persistent components) + +**Active windows** — in-flight, currently being written to: + +| Ingest type | `Mem_active(g)` | +|-------------|----------------| +| Tumbling | `N(s,g) × mem(s,p)` | +| Sliding | `⌈W/S⌉ × N(s,g) × mem(s,p)` | + +Sliding has `⌈W/S⌉` concurrent overlapping windows open at any moment; +each arriving item is inserted into all of them. + +**Retained windows** — completed windows kept for query lookback: + +| Ingest type | `Mem_retain(g)` | +|-------------|----------------| +| Tumbling | `n × N(s,g) × mem(s,p)` | +| Sliding | `N(s,g) × mem(s,p)` (W = range_a forced, so n = 1) | + +Note: for subtractable sketches (tumbling only), one additional running +cumulative prefix instance is maintained in active memory: +`+N(s,g) × mem(s,p)` added to `Mem_active`. + +**Total ingestion memory:** +``` +Mem_ingest(g) = Mem_active(g) + Mem_retain(g) +``` + +#### Ingestion CPU (rate — cost per second) + +| Ingest type | `CPU_ingest(g)` | +|-------------|----------------| +| Tumbling | `ρ_g × insert_cpu(s,p)` | +| Sliding | `ρ_g × ⌈W/S⌉ × insert_cpu(s,p)` | + +Does **not** scale with `N(s,g)`: each arriving item triggers exactly one +insert regardless of subpopulation structure (routing overhead ignored per +modelling assumption). + +#### Query CPU (cost per query invocation) + +| Query method | `CPU_query(a,g)` | +|--------------|----------------| +| Neither | `N(s,g) × query_cpu(s,p)` | +| Merge | `N(s,g) × ((n - 1) × merge_cpu(s,p) + query_cpu(s,p))` | +| Subtract | `N(s,g) × (subtract_cpu(s,p) + query_cpu(s,p))` | + +For merge: `n - 1` pairwise merge operations reduce `n` retained windows to +one, then one query on the result. Subtract is O(1) with respect to `n`. + +#### Query memory (transient — peak working memory per query invocation) + +| Query method | `Mem_query(a,g)` | +|--------------|----------------| +| Neither | `N(s,g) × mem(s,p)` | +| Merge | `n × N(s,g) × mem(s,p)` (reducible to `2 × N(s,g) × mem(s,p)` with incremental merge) | +| Subtract | `2 × N(s,g) × mem(s,p)` (two prefix checkpoints loaded simultaneously) | + +Query memory is transient (not steady-state) and is not amortised across +concurrent queries (concurrent queries are out of scope for v1). + +### Connection to the MIP objective + +``` +IngestCost(g) = w₁ × CPU_ingest(g) + w₂ × Mem_ingest(g) +QueryCost(a,g) = w₃ × CPU_query(a,g) + w₄ × Mem_query(a,g) +``` + +Both are now fully concrete functions of `g` and the profiled inputs +`(θ_a, ρ_g, range_a)`. The query method (neither / merge / subtract) is +read off the compatibility table above given `g`'s ingest type, `W`, and +`s`'s algebraic properties — so `QueryCost(a,g)` remains assignment-independent. + +## Decision variables + +- `y_g ∈ {0,1}` — is config `g` deployed. +- `x_{a,g} ∈ {0,1}`, defined only where `Feasible(a,g)=1` — is AQE `a` + served by config `g`. + +## Objective + +``` +minimize Σ_g y_g · IngestCost(g) + Σ_{a,g} x_{a,g} · f_a · QueryCost(a,g) +``` + +Both terms are cost-per-second; `f_a` converts the per-query `QueryCost` +into a rate so it's commensurate with the continuously-accruing +`IngestCost`. + +## Constraints + +``` +Σ_{g: Feasible(a,g)} x_{a,g} = 1 for all a ∈ A (each AQE assigned to exactly one config) +x_{a,g} ≤ y_g for all a, g (can't use an undeployed config) +x_{a,g}, y_g ∈ {0,1} +``` + +**Optional latency constraint** (v1: off by default). For AQEs that +specify an `SLA_a` (otherwise `SLA_a = ∞`, i.e. unconstrained): + +``` +x_{a,g} = 1 ⟹ Latency_query(a,g) ≤ SLA_a +``` + +equivalently, fold this into `Feasible(a,g)` as a fifth condition +(`Feasible(a,g) := ... AND Latency_query(a,g) ≤ SLA_a`), keeping the same +"infeasible pairs are filtered out before the MIP runs" pattern used for +accuracy, retention, and ingest/query compatibility. + +Because `EXACT_a` is always feasible for `a`, the assignment constraint is +always satisfiable — there's no need for a separate "is this RQE supported" +boolean or a hard-coverage constraint; "unsupported" simply means the +optimizer picked `EXACT_a` because no sketch config met `ε_a` more cheaply. + +## Problem class + +This is an **uncapacitated facility-location MIP**: `y_g` = "open facility +g", `x_{a,g}` = "assign demand point a to facility g", `IngestCost(g)` = +facility's fixed cost, `f_a·QueryCost(a,g)` = assignment cost. Standard +structure — solvable via off-the-shelf MIP solvers (CBC/OR-tools/Gurobi) or +LP-relaxation + rounding heuristics. Expected problem size (AQEs × candidate +configs in the hundreds) should be comfortably within reach of exact +solvers. No wall-clock solve-time budget imposed on v1 — the optimizer runs +as an offline planning step before deployment, so any solve time is +acceptable. + +## Generating `G` + +`G` is not given — it must be enumerated: + +1. Per AQE `a`, propose candidates from sketch types compatible with `a`'s + statistic type (today's `map_statistic_to_precompute_operator` + compatibility list) × sketch-bench's parameter grid × a window-size grid + (multiples of scrape interval) × a retention-depth grid (multiples of + `range_a / window_size`, since deeper retention only matters if some AQE + needs that much range). +2. For sharing across AQEs, check pairs/groups with the same metric and a + label-superset relationship: if one AQE's labels are a superset of + another's and windows align, a single rolled-up config can serve both — + this is the `CAPABILITY_MATCHING_DESIGN.md` predicate, reused as + `Feasible`. +3. `θ_a` profiling must run before `Error(a,g,θ_a)` can be computed for any + `g` — it's a precursor pass, not part of the MIP. + +For v1, assume this enumeration is combinatorially manageable (no pruning +heuristic designed yet — revisit if `|G|` turns out too large for the +solver in practice). + +**Candidate generation is reactive, not proactive.** For each metric, only +propose configs at label granularities that existing AQEs in the current +optimization run actually need — do not enumerate finer label combinations +than any AQE requires. A finer-grained config that no current AQE benefits +from only adds ingest cost with no benefit in a static one-shot problem. +(Cross-AQE sharing via step 2 still happens: if AQE `a1` needs `{region}` +and AQE `a2` needs `{region, instance}`, `a2`'s generated config at +`{region, instance}` is also checked as a candidate for `a1`, since +`Feasible(a1, g_{region,instance}) = true` via merge.) + +### Data gap: sketch-bench doesn't yet have the data this formulation needs + +Investigated `sketch-bench`'s actual schema and sweep behavior; two findings +that block using it as-is for `Error(a,g,θ_a)` lookups: + +- **Cardinality is not a swept parameter.** `sketch-cli/src/main.rs:85-87` + takes a single `--cardinality` value per invocation (default 100,000); + `docs/BENCH_SWEEP.md`'s default grids only vary sketch params (e.g. CMS + `rows×cols`, HLL `lg_k`), never cardinality. A real output record looks + like: + ```json + {"sketch_config": {"family": "cms", "params": {"cols": 1024, "rows": 3}}, + "workload": {"shape": "zipf", "size": 1000000, "cardinality": 100000, ...}, + "bench": {"accuracy": {"relative_error_mean": 13.438, ...}}} + ``` + `cardinality` is fixed per run, not a list. To get accuracy at multiple + cardinalities (needed to look up `Error` at whatever effective cardinality + a given `window_size` implies), someone has to add cardinality to the + sweep grid and re-run the benchmarks — this data doesn't exist yet. +- **No windowing concept at all.** The only "window" in `sketch-core` + (`report.rs:17`) means measurement-repetition window (N runs averaged), + not a stream/tumbling window. Workloads are loaded as one static + in-memory batch, then queried (README.md:120). The formulation's + assumption — "window size determines the effective cardinality the + sketch must absorb before flush, so look that up against sketch-bench's + cardinality axis" — translates streaming semantics onto sketch-bench's + batch semantics. That translation is plausible but unverified: it assumes + a tumbling window's insert pattern is well-approximated by sketch-bench's + batch-load-then-query pattern, which sketch-bench was never designed to + validate. + +Until sketch-bench's sweep grid includes cardinality and someone confirms +the batch-vs-window translation is reasonable, analytic error bounds (where +they exist) are the safer source for `Error(a,g,θ_a)` rather than empirical +lookup. + +Extending sketch-bench (adding cardinality to the sweep grids, validating +the batch-vs-window translation) is in scope for this project, not punted +to a separate workstream. + +### Data gap: heap-based sketches have cardinality-dependent memory + +`CountMinSketchWithHeap` (`asap-internal/sketch-core/src/count_min_with_heap.rs:46-67`) +holds its top-k entries in a `Vec` that starts empty and grows on +insert until it reaches `heap_size`; each `HeapItem` stores a `String` key, +so memory depends on the number of distinct keys actually seen (up to the +cap) and on key length — not purely on configured params. (`HydraKLL` is +fine: a fixed-size grid of fixed-capacity KLL buffers, memory determined +entirely by `params`.) This is the same structural risk retention posed: +`Mem_ingest(g)` would depend on the assignment (how many distinct keys the +assigned AQEs' data actually has), not just on `g`. + +`CountMinSketchWithHeap` is also not currently benchmarked in sketch-bench +at all (`sketch-cli/src/wrappers/cms.rs` only wraps non-heap CMS variants: +`oxide`, `datasketches`, `lib_vector2d_*`) — another item to add when +extending sketch-bench. + +Fix, analogous to retention: use the **worst-case (saturated) memory +bound** — `heap_size · avg_key_size` — as `Mem_ingest(g)` for any config +using this sketch, rather than measuring actual usage. This keeps +`IngestCost(g)` assignment-independent (conservative when real cardinality +is below `heap_size`, exact once the heap saturates, which it will for any +AQE with cardinality ≥ `heap_size`). + +## Out of scope for v1 (explicitly punted) + +- **Hard resource budget constraints** (e.g. `Σ_g y_g·Mem_ingest(g) ≤ + M_total`). v1 only weights memory/CPU into the objective via `w₂`/`w₄`; + nothing stops a solution from exceeding real cluster capacity. Add as a + hard constraint later if this becomes a problem in practice. +- **Engine limitations** (e.g. Arroyo can't do sliding windows today) are + treated purely as a filter on `G`'s generation (step 1 above), not as part + of the math itself, so the formulation stays engine-agnostic as Arroyo's + capabilities change. +- **Decomposing `OneTemporalOneSpatial` AQEs into separate temporal/spatial + pipeline stages.** Today (and in this v1 formulation) a combined + temporal+spatial AQE like `sum(quantile_over_time(0.9, metric[5m]))` + (`planner/patterns.rs:75-103`) is one atomic unit mapped to one `g`. This + hides sharing: two RQEs differing only in their spatial reducer (e.g. + `sum(...)` vs `avg(...)` over the same temporal sub-computation) can't + share the temporal stage if `g` stays atomic. Decomposing into a + multi-stage pipeline (`g` as a DAG, not a single config) would expose + this reuse but adds real modeling complexity (`Feasible`/`IngestCost` + need to handle chained stages). Keep AQEs atomic for v1; revisit if this + sharing turns out to matter in practice. +- **Workload drift / re-planning.** v1 assumes a static, one-shot + optimization over a fixed, known set of RQEs. No switching-cost term for + redeploying configs when the workload changes, and no online/adaptive + profiling of `θ_a` (it's assumed measured once, upfront). Revisit once the + system needs to handle RQE sets that change over time. +- **Weight calibration** (`w₁..w₄`). No method is specified yet for how + these get chosen (e.g. fixed $/CPU-sec, $/GB-sec conversions, or some + other tuning process). The math is agnostic to their values; calibrating + them is a separate problem. +- **Global weights, not per-AQE/per-RQE.** v1 assumes one global + `(w₁,w₂,w₃,w₄)` vector for the whole system, not per-AQE priorities (e.g. + an on-call dashboard weighted more latency-sensitive than a weekly + report). This is a deliberate v1 simplification, not a discovered + constraint — revisit if different RQEs need genuinely different + cost/accuracy tradeoffs. +- **`θ_a` profiling/estimation step itself.** This doc assumes `θ_a` is + available as input; how to actually estimate real cardinality/skew per + AQE (sampling strategy, accuracy of the estimate, cost of running it) is + not designed here. +- **Binary-op evaluation cost.** Combining two AQE results at query time + (e.g. the division in a quantile-ratio QE) is assumed negligible compared + to sketch query cost and isn't given its own cost term. Revisit if this + assumption turns out false for some operator/sketch combination. +- **Pruning `G`'s enumeration.** Assumed combinatorially manageable for v1 + with no pruning strategy designed. If sketch type × param grid × window + size × retention depth × sharing candidates turns out too large for the + solver, will need a pruning heuristic (e.g. only propose Pareto-optimal + `(cost, accuracy)` configs per AQE) — not designed yet. +- **Extending sketch-bench to cover the cardinality/window axis** this + formulation needs (see "Data gap" above) — adding cardinality to the + default sweep grids, and validating that batch-load benchmarks transfer + to tumbling-window insert patterns. +- **Sketch property catalogue** — which specific sketch families are + `mergeable`, `subtractable`, `subpopulation_aware`. The analytical cost + model uses these properties to determine valid query methods and cost + scaling, but the actual per-family values are not yet catalogued. To be + defined once all sketch implementations are reviewed (includes confirming + whether `CountMinSketchWithHeap` is subtractable via its CMS matrix alone, + whether `HydraKLL` is mergeable via element-wise KLL cell merging, etc.). + +## Where this would slot into the code (not yet implemented) + +- `planner/sketch.rs`'s hardcoded `Statistic → AggregationType` map and + fixed parameter constants would be replaced by step 1 of `G`'s generation + plus the MIP's `x_{a,g}` assignment. +- `planner/window.rs`'s hardcoded window sizing would become part of `G`'s + window-size/retention-depth grid instead of a fixed formula. +- `CAPABILITY_MATCHING_DESIGN.md`'s capability-matching logic becomes the + `Feasible(a,g)` predicate (extended with the accuracy and retention + checks above). +- A new component is needed to: (a) run the `θ_a` profiling pass, (b) query + sketch-bench data (or an analytic formula) for `IngestCost`, `QueryCost`, + `Error`, (c) enumerate `G`, (d) invoke a MIP solver, (e) translate the + `x_{a,g}`/`y_g` solution into `StreamingConfig`/`InferenceConfig` structs. diff --git a/asap-planner-rs/Cargo.toml b/asap-planner-rs/Cargo.toml index 53a32fc7..99544db0 100644 --- a/asap-planner-rs/Cargo.toml +++ b/asap-planner-rs/Cargo.toml @@ -11,6 +11,10 @@ path = "src/lib.rs" name = "asap-planner" path = "src/main.rs" +[[bin]] +name = "asap-optimizer-cli" +path = "src/bin/optimizer_cli.rs" + [dependencies] asap_types.workspace = true promql_utilities.workspace = true diff --git a/asap-planner-rs/Dockerfile b/asap-planner-rs/Dockerfile index ca7b183f..6da5ede7 100644 --- a/asap-planner-rs/Dockerfile +++ b/asap-planner-rs/Dockerfile @@ -19,7 +19,8 @@ COPY asap-planner-rs/Cargo.toml ./asap-planner-rs/ # Create dummy source files so Cargo can resolve all workspace members RUN mkdir -p asap-query-engine/src && echo "fn main() {}" > asap-query-engine/src/main.rs && \ mkdir -p asap-query-engine/benches && echo "fn main() {}" > asap-query-engine/benches/simple_store_bench.rs && \ - mkdir -p asap-planner-rs/src && echo "fn main() {}" > asap-planner-rs/src/main.rs && \ + mkdir -p asap-planner-rs/src/bin && echo "fn main() {}" > asap-planner-rs/src/main.rs && \ + echo "fn main() {}" > asap-planner-rs/src/bin/optimizer_cli.rs && \ echo "pub fn placeholder() {}" >> asap-planner-rs/src/lib.rs # Build dependencies (this layer will be cached) diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs new file mode 100644 index 00000000..02eec53c --- /dev/null +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -0,0 +1,83 @@ +//! Offline runner for the optimization-based sketch/config selector. +//! +//! Standalone: not wired into `asap-planner`/`Controller::generate()` yet. Lets +//! you exercise `run_greedy_pipeline` against real workload configs while the +//! optimizer module is still under development (Phase 2 of issue #405). + +use std::path::PathBuf; + +use asap_planner::optimizer::run_greedy_pipeline; +use asap_planner::ControllerConfig; +use clap::Parser; + +#[derive(Parser, Debug)] +#[command( + name = "asap-optimizer-cli", + about = "Offline runner for the optimization-based sketch/config selector (not wired into asap-planner yet)" +)] +struct Args { + /// Path to a YAML workload config (same format as `asap-planner --input_config`). + #[arg(long = "input_config")] + input_config: PathBuf, + + #[arg(long = "prometheus_scrape_interval")] + prometheus_scrape_interval: 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 + /// in .design_docs/optimizer-v1-implementation-plan.md. + #[arg(long = "rho", default_value = "1.0", value_parser = parse_positive_finite)] + rho: f64, + + #[arg(short, long, action = clap::ArgAction::Count)] + verbose: u8, +} + +fn parse_positive_finite(s: &str) -> Result { + let v: f64 = s.parse().map_err(|_| format!("not a valid number: {s}"))?; + if !v.is_finite() || v <= 0.0 { + return Err(format!("--rho must be a positive finite number, got {v}")); + } + Ok(v) +} + +fn main() -> anyhow::Result<()> { + let args = Args::parse(); + + tracing_subscriber::fmt() + .with_max_level(if args.verbose > 0 { + tracing::Level::DEBUG + } else { + tracing::Level::INFO + }) + .init(); + + let yaml_str = std::fs::read_to_string(&args.input_config)?; + 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, args.rho); + + let deployed = streaming.get_all_aggregation_configs(); + println!("=== Deployed streaming configs: {} ===", deployed.len()); + for (id, cfg) in deployed { + println!( + " [{id}] {} sub_type={:?} window={}s slide={}s type={:?} metric={} params={:?}", + cfg.aggregation_type, + cfg.aggregation_sub_type, + cfg.window_size, + cfg.slide_interval, + cfg.window_type, + cfg.metric, + cfg.parameters, + ); + } + + println!("\n=== Query configs: {} ===", inference.query_configs.len()); + for qc in &inference.query_configs { + println!(" \"{}\" -> {:?}", qc.query, qc.aggregations); + } + + Ok(()) +} diff --git a/asap-planner-rs/src/lib.rs b/asap-planner-rs/src/lib.rs index 40ca3bf9..f6d691ea 100644 --- a/asap-planner-rs/src/lib.rs +++ b/asap-planner-rs/src/lib.rs @@ -3,6 +3,7 @@ pub mod config; pub mod elastic_dsl; pub mod error; pub mod generator; +pub mod optimizer; pub mod planner; pub mod planner_output; pub mod prometheus_client; diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs new file mode 100644 index 00000000..d9032421 --- /dev/null +++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs @@ -0,0 +1,310 @@ +use std::collections::HashMap; + +use asap_types::query_requirements::QueryRequirements; +use asap_types::utils::normalize_spatial_filter; +use asap_types::PromQLSchema; +use promql_utilities::data_model::KeyByLabelNames; +use promql_utilities::query_logics::enums::{QueryPatternType, Statistic}; +use promql_utilities::query_logics::parsing::{ + get_metric_and_spatial_filter, get_spatial_aggregation_output_labels, get_statistics_to_compute, +}; +use tracing::warn; + +use crate::planner::patterns::build_patterns; + +use super::solution::AQE; + +/// One repeating query expression: a PromQL query string and its repetition +/// interval (e.g. the refresh interval of the dashboard panel it belongs to). +#[derive(Debug, Clone)] +pub struct RQE { + pub query_string: String, + pub t_repeat_secs: u64, +} + +/// Stable deduplication key for an AQE. +/// Two leaf queries that produce identical requirements are treated as the same +/// AQE regardless of which RQE they came from. +#[derive(Debug, Clone, Hash, PartialEq, Eq)] +struct AQEKey { + metric: String, + /// Statistics are produced in a stable order by get_statistics_to_compute. + statistics: Vec, + data_range_ms: Option, + grouping_labels: KeyByLabelNames, + spatial_filter_normalized: String, + topk_count_events: Option, +} + +impl AQEKey { + fn from_requirements(req: &QueryRequirements) -> Self { + Self { + metric: req.metric.clone(), + statistics: req.statistics.clone(), + data_range_ms: req.data_range_ms, + grouping_labels: req.grouping_labels.clone(), + spatial_filter_normalized: req.spatial_filter_normalized.clone(), + topk_count_events: req.topk_count_events, + } + } +} + +/// Extract and deduplicate AQEs from a set of RQEs. +/// +/// Each RQE is decomposed into leaf query expressions (recursively splitting +/// binary arithmetic operators), then each leaf is pattern-matched to produce +/// a `QueryRequirements`. AQEs with identical requirements are merged. +/// +/// Three frequency-related values are computed per AQE: +/// - `query_frequency_hz`: Σ 1/T_r — total query load for the MIP objective. +/// - `min_t_repeat_secs`: min(T_r) — freshness bound on window size W ≤ min_t. +/// - `t_repeat_gcd_secs`: GCD(T_r) — natural slide interval S for candidate +/// generation (windows completing every GCD secs align with all dashboards). +/// +/// Leaf queries that do not match any supported pattern (e.g. unsupported +/// functions, parse errors) are skipped with a warning. +pub fn extract_aqes(rqes: &[RQE], metric_schema: &PromQLSchema) -> Vec { + // (key) -> (requirements, query_strings, sum_freq, min_t, gcd_t) + let mut acc: HashMap, f64, u64, u64)> = HashMap::new(); + + for rqe in rqes { + if rqe.t_repeat_secs == 0 { + warn!( + query = %rqe.query_string, + "aqe_extractor: skipping RQE with repetition_delay=0 \ + (would produce infinite query frequency and corrupt GCD)" + ); + continue; + } + + let leaves = decompose_to_leaves(&rqe.query_string); + + for leaf in leaves { + match extract_requirements(&leaf, metric_schema) { + Some(req) => { + let key = AQEKey::from_requirements(&req); + let entry = acc + .entry(key) + .or_insert_with(|| (req, Vec::new(), 0.0, u64::MAX, 0)); + if !entry.1.contains(&leaf) { + entry.1.push(leaf); + } + entry.2 += 1.0 / rqe.t_repeat_secs as f64; + entry.3 = entry.3.min(rqe.t_repeat_secs); + entry.4 = if entry.4 == 0 { + rqe.t_repeat_secs + } else { + gcd(entry.4, rqe.t_repeat_secs) + }; + } + None => { + warn!( + query = %leaf, + "aqe_extractor: skipping unsupported or unparseable leaf query" + ); + } + } + } + } + + acc.into_values() + .map( + |( + requirements, + query_strings, + query_frequency_hz, + min_t_repeat_secs, + t_repeat_gcd_secs, + )| AQE { + requirements, + query_strings, + query_frequency_hz, + min_t_repeat_secs, + t_repeat_gcd_secs, + }, + ) + .collect() +} + +/// Euclidean GCD. `num-integer` is not in the workspace; this two-liner is +/// sufficient and avoids a dependency. +fn gcd(a: u64, b: u64) -> u64 { + if b == 0 { + a + } else { + gcd(b, a % b) + } +} + +/// Recursively decompose a PromQL expression into non-binary leaf queries. +/// +/// Binary arithmetic expressions (e.g. `rate(a[5m]) / rate(b[5m])`) are split +/// into their arms. Scalar arms (e.g. the `100` in `rate(x[5m]) * 100`) are +/// dropped — they contribute no AQE. Only arithmetic operators are split; +/// comparison and set operators are left as-is (treated as opaque leaves). +fn decompose_to_leaves(query: &str) -> Vec { + let ast = match promql_parser::parser::parse(query) { + Ok(a) => a, + Err(_) => return vec![query.to_string()], + }; + + if let promql_parser::parser::Expr::Binary(binary) = &ast { + if !binary.op.is_comparison_operator() && !binary.op.is_set_operator() { + let mut leaves = Vec::new(); + if let Some(lhs_str) = arm_to_query_string(binary.lhs.as_ref()) { + leaves.extend(decompose_to_leaves(&lhs_str)); + } + if let Some(rhs_str) = arm_to_query_string(binary.rhs.as_ref()) { + leaves.extend(decompose_to_leaves(&rhs_str)); + } + return leaves; + } + } + + vec![query.to_string()] +} + +/// Convert one arm of a binary expression to a query string, returning `None` +/// for scalar literals (they don't map to AQEs). +fn arm_to_query_string(expr: &promql_parser::parser::Expr) -> Option { + let inner = strip_parens(expr); + match inner { + promql_parser::parser::Expr::NumberLiteral(_) => None, + other => Some(format!("{}", other)), + } +} + +fn strip_parens(expr: &promql_parser::parser::Expr) -> &promql_parser::parser::Expr { + if let promql_parser::parser::Expr::Paren(paren) = expr { + strip_parens(&paren.expr) + } else { + expr + } +} + +/// Try to extract `QueryRequirements` from a single leaf PromQL query string. +/// Returns `None` if the query cannot be parsed or does not match any pattern. +/// +/// TODO: this duplicates `build_query_requirements_promql` in +/// `asap-query-engine/src/engines/simple_engine/promql.rs:614`. That function +/// is a private `&self` method tied to `SimplePromQLEngine`. The shared logic +/// should be extracted into a free function in `asap_types::query_requirements` +/// and called from both sites. +fn extract_requirements(query: &str, metric_schema: &PromQLSchema) -> Option { + let ast = promql_parser::parser::parse(query).ok()?; + let patterns = build_patterns(); + + let (pattern_type, match_result) = patterns.iter().find_map(|(pt, pat)| { + let r = pat.matches(&ast); + if r.matches { + Some((*pt, r)) + } else { + None + } + })?; + + let (metric, spatial_filter) = get_metric_and_spatial_filter(&match_result); + let statistics = get_statistics_to_compute(pattern_type, &match_result); + + let data_range_ms = match pattern_type { + QueryPatternType::OnlySpatial => None, + _ => match_result + .get_range_duration() + .map(|d| d.num_seconds() as u64 * 1000), + }; + + let grouping_labels = match pattern_type { + // OnlyTemporal preserves all labels — look them up in the schema. + // If the metric is unknown, fall back to empty (dedup still works; cost + // model will treat it as a zero-group-count sketch). + QueryPatternType::OnlyTemporal => metric_schema + .get_labels(&metric) + .cloned() + .unwrap_or_else(KeyByLabelNames::empty), + // OnlySpatial and OneTemporalOneSpatial encode their output labels in + // the AST's `by (...)` / `without (...)` clause. + QueryPatternType::OnlySpatial | QueryPatternType::OneTemporalOneSpatial => { + let all_labels = metric_schema + .get_labels(&metric) + .cloned() + .unwrap_or_else(KeyByLabelNames::empty); + get_spatial_aggregation_output_labels(&match_result, &all_labels) + } + }; + + Some(QueryRequirements { + metric, + statistics, + data_range_ms, + grouping_labels, + spatial_filter_normalized: normalize_spatial_filter(&spatial_filter), + topk_count_events: None, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn empty_schema() -> PromQLSchema { + PromQLSchema::new() + } + + fn rqe(query: &str, t: u64) -> RQE { + RQE { + query_string: query.to_string(), + t_repeat_secs: t, + } + } + + #[test] + fn single_temporal_query() { + let rqes = vec![rqe("sum_over_time(metric[5m])", 60)]; + let aqes = extract_aqes(&rqes, &empty_schema()); + assert_eq!(aqes.len(), 1); + assert!((aqes[0].query_frequency_hz - 1.0 / 60.0).abs() < 1e-9); + assert_eq!(aqes[0].requirements.metric, "metric"); + assert_eq!(aqes[0].requirements.data_range_ms, Some(300_000)); + } + + #[test] + fn binary_query_produces_two_aqes() { + let rqes = vec![rqe( + "sum_over_time(metric_a[5m]) / sum_over_time(metric_b[5m])", + 60, + )]; + let aqes = extract_aqes(&rqes, &empty_schema()); + assert_eq!(aqes.len(), 2); + } + + #[test] + fn binary_with_scalar_produces_one_aqe() { + let rqes = vec![rqe("sum_over_time(metric[5m]) * 100", 60)]; + let aqes = extract_aqes(&rqes, &empty_schema()); + assert_eq!(aqes.len(), 1); + } + + #[test] + fn same_aqe_in_two_rqes_deduplicates_and_sums_frequency() { + let rqes = vec![ + rqe("sum_over_time(metric[5m])", 60), + rqe("sum_over_time(metric[5m])", 30), + ]; + let aqes = extract_aqes(&rqes, &empty_schema()); + assert_eq!(aqes.len(), 1); + // f_a = sum of rates (total query load for the MIP objective) + let expected_freq = 1.0 / 60.0 + 1.0 / 30.0; + assert!((aqes[0].query_frequency_hz - expected_freq).abs() < 1e-9); + // min_t and gcd_t used for windowing constraints + assert_eq!(aqes[0].min_t_repeat_secs, 30); + assert_eq!(aqes[0].t_repeat_gcd_secs, 30); // gcd(60, 30) = 30 + assert_eq!(aqes[0].query_strings.len(), 1); // same string, deduplicated + } + + #[test] + fn unsupported_query_is_skipped() { + let rqes = vec![rqe("not_a_real_function(metric[5m])", 60)]; + let aqes = extract_aqes(&rqes, &empty_schema()); + assert_eq!(aqes.len(), 0); + } +} diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs new file mode 100644 index 00000000..c294841d --- /dev/null +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -0,0 +1,380 @@ +use std::collections::HashMap; + +use asap_types::aggregation_config::AggregationConfig; +use asap_types::capability_matching::compatible_agg_types; +use asap_types::enums::WindowType; +use promql_utilities::data_model::KeyByLabelNames; +use promql_utilities::query_logics::enums::{AggregationType, Statistic}; +use serde_json::Value; + +use super::constants::{ + CMS_DEPTHS, CMS_HEAP_SIZES, CMS_WIDTHS, HLL_PRECISIONS, HYDRA_COLS, HYDRA_K, HYDRA_ROWS, KLL_KS, +}; +use super::sketch_properties::sketch_properties; +use super::solution::{QueryMethod, AQE}; + +/// A candidate streaming config for a single AQE, ready for cost evaluation. +#[derive(Debug, Clone)] +pub struct CandidateConfig { + /// None = EXACT fallback (no streaming config; raw Prometheus query at query time). + pub config: Option, + /// Query method derived from (ingest type × W vs range_a × sketch algebra). + pub query_method: QueryMethod, + /// Number of retained windows used at query time (n for Merge, 1 for Direct/Subtract, 0 for Exact). + pub n_windows: u64, +} + +/// Enumerate all candidate configs for an AQE. +/// +/// Iterates over compatible agg types × parameter grid × valid window sizes × +/// {Tumbling, Sliding}. Always appends an EXACT candidate last (always feasible). +/// +/// Multi-statistic AQEs (e.g. avg = [Sum, Count]) return only EXACT — a single +/// sketch family cannot serve two incompatible statistics simultaneously. +pub fn enumerate_candidates(aqe: &AQE, scrape_interval_secs: u64) -> Vec { + let mut candidates = Vec::new(); + + if aqe.requirements.statistics.len() != 1 { + // ponytail: multi-stat AQEs (avg) need two sketches; not supported in v1. + candidates.push(exact_candidate()); + return candidates; + } + + let stat = aqe.requirements.statistics[0]; + let range_a_secs = aqe.requirements.data_range_ms.map(|ms| ms.div_ceil(1000)); + + for &agg_type in compatible_agg_types(stat) { + let props = sketch_properties(agg_type); + let sub_type = derive_sub_type(stat, agg_type); + + for params in param_grid(agg_type, aqe.requirements.topk_count_events) { + for (window_type, w, slide_interval, n) in + window_candidates(range_a_secs, aqe.min_t_repeat_secs, scrape_interval_secs) + { + let Some(qm) = determine_query_method(window_type, n, &props) else { + continue; + }; + + let config = build_config( + aqe, + agg_type, + &sub_type, + ¶ms, + window_type, + w, + slide_interval, + n, + ); + candidates.push(CandidateConfig { + config: Some(config), + query_method: qm, + n_windows: n, + }); + } + } + } + + candidates.push(exact_candidate()); + candidates +} + +fn exact_candidate() -> CandidateConfig { + CandidateConfig { + config: None, + query_method: QueryMethod::Exact, + n_windows: 0, + } +} + +/// Window candidates: (WindowType, W_secs, slide_interval_secs, n_windows). +/// +/// For spatial-only queries (range = None): one tumbling window of scrape_interval. +/// For range-based queries: +/// Tumbling: all W that divide range_a, are multiples of scrape_interval, and ≤ min_t_repeat. +/// Slide interval = W (a tumbling window "slides" by its own width). +/// Sliding: W = range_a exactly (overlapping sliding windows can't merge/subtract to cover +/// a larger range — only mutually-exclusive tumbling windows compose). The slide +/// interval S is still a free choice ≤ W: smaller S means more concurrent +/// overlapping sub-windows maintained internally (⌈W/S⌉), which costs more +/// ingest CPU/mem but gives fresher results. Doubling from scrape_interval up to +/// W keeps the grid small while covering that tradeoff. +fn window_candidates( + range_a_secs: Option, + min_t_repeat_secs: u64, + scrape_interval_secs: u64, +) -> Vec<(WindowType, u64, u64, u64)> { + let Some(range_a) = range_a_secs else { + // Spatial-only: any window works (window_compatible returns true for None range). + return vec![( + WindowType::Tumbling, + scrape_interval_secs, + scrape_interval_secs, + 1, + )]; + }; + if range_a == 0 || scrape_interval_secs == 0 { + return vec![]; + } + + let max_w = range_a.min(min_t_repeat_secs); + let mut out = Vec::new(); + + // Tumbling: W divides range_a, is a multiple of scrape_interval, and ≤ max_w. + let mut w = scrape_interval_secs; + while w <= max_w { + if range_a % w == 0 { + let n = range_a / w; + out.push((WindowType::Tumbling, w, w, n)); + } + w += scrape_interval_secs; + } + + // Sliding: W = range_a exactly; S doubles from scrape_interval up to W. + if range_a <= min_t_repeat_secs { + let mut s = scrape_interval_secs; + while s <= range_a { + out.push((WindowType::Sliding, range_a, s, 1)); + s *= 2; + } + } + + out +} + +/// Determine query method from (window_type, n_windows, sketch algebra). +/// Returns None when the combination is infeasible (tumbling + W < range_a + neither property). +fn determine_query_method( + window_type: WindowType, + n_windows: u64, + props: &super::sketch_properties::SketchProperties, +) -> Option { + if n_windows == 1 { + // W = range_a (or spatial-only): one completed window covers the query range exactly. + return Some(QueryMethod::Direct); + } + // n > 1 means W < range_a; sliding is excluded (sliding enforces W = range_a above). + debug_assert_eq!(window_type, WindowType::Tumbling); + if props.subtractable { + Some(QueryMethod::Subtract) + } else if props.mergeable { + debug_assert!(n_windows > 1, "Merge requires merging >1 window"); + Some(QueryMethod::Merge { + num_windows: n_windows, + }) + } else { + None + } +} + +/// Build an AggregationConfig from candidate parameters. aggregation_id = 0 (placeholder). +#[allow(clippy::too_many_arguments)] +fn build_config( + aqe: &AQE, + agg_type: AggregationType, + sub_type: &str, + params: &HashMap, + window_type: WindowType, + w: u64, + slide_interval: u64, + n_windows: u64, +) -> AggregationConfig { + AggregationConfig::new( + 0, // placeholder; replaced by greedy/MIP solver when deploying + agg_type, + sub_type.to_string(), + params.clone(), + aqe.requirements.grouping_labels.clone(), + KeyByLabelNames::empty(), // aggregated_labels (not needed for optimizer feasibility) + KeyByLabelNames::empty(), // rollup_labels + String::new(), // original_yaml + w, + slide_interval, + window_type, + aqe.requirements.spatial_filter_normalized.clone(), + aqe.requirements.metric.clone(), + Some(n_windows), + None, // read_count_threshold + None, // table_name (SQL only) + None, // value_column (SQL only) + ) +} + +/// aggregation_sub_type string expected by the streaming engine and capability matching. +fn derive_sub_type(stat: Statistic, agg_type: AggregationType) -> String { + match (stat, agg_type) { + (Statistic::Min, _) => "min", + (Statistic::Max, _) => "max", + (Statistic::Topk, _) => "topk", + (Statistic::Sum, AggregationType::CountMinSketch) => "sum", + (Statistic::Count, AggregationType::CountMinSketch) => "count", + _ => "", + } + .to_string() +} + +fn param_grid( + agg_type: AggregationType, + topk_count_events: Option, +) -> Vec> { + match agg_type { + AggregationType::CountMinSketch => { + let mut grids = Vec::new(); + for &d in CMS_DEPTHS { + for &w in CMS_WIDTHS { + let mut m = HashMap::new(); + m.insert("depth".into(), Value::from(d)); + m.insert("width".into(), Value::from(w)); + grids.push(m); + } + } + grids + } + + AggregationType::CountMinSketchWithHeap => { + let count_events_variants: &[bool] = match topk_count_events { + Some(v) => { + if v { + &[true] + } else { + &[false] + } + } + None => &[true, false], + }; + let mut grids = Vec::new(); + for &d in CMS_DEPTHS { + for &w in CMS_WIDTHS { + for &h in CMS_HEAP_SIZES { + for &ce in count_events_variants { + let mut m = HashMap::new(); + m.insert("depth".into(), Value::from(d)); + m.insert("width".into(), Value::from(w)); + m.insert("heapsize".into(), Value::from(h)); + m.insert("count_events".into(), Value::from(ce)); + grids.push(m); + } + } + } + } + grids + } + + AggregationType::DatasketchesKLL => KLL_KS + .iter() + .map(|&k| { + let mut m = HashMap::new(); + m.insert("K".into(), Value::from(k)); + m + }) + .collect(), + + AggregationType::HydraKLL => { + let mut grids = Vec::new(); + for &r in HYDRA_ROWS { + for &c in HYDRA_COLS { + let mut m = HashMap::new(); + m.insert("row_num".into(), Value::from(r)); + m.insert("col_num".into(), Value::from(c)); + m.insert("k".into(), Value::from(HYDRA_K)); + grids.push(m); + } + } + grids + } + + AggregationType::HLL => HLL_PRECISIONS + .iter() + .map(|&p| { + let mut m = HashMap::new(); + m.insert("precision".into(), Value::from(p)); + m + }) + .collect(), + + // Parameterless types: one empty-params entry per type. + _ => vec![HashMap::new()], + } +} + +#[cfg(test)] +mod tests { + use super::*; + use promql_utilities::data_model::KeyByLabelNames; + + fn make_aqe(stat: Statistic, range_ms: Option, min_t: u64) -> AQE { + use asap_types::query_requirements::QueryRequirements; + AQE { + requirements: QueryRequirements { + metric: "test_metric".into(), + statistics: vec![stat], + data_range_ms: range_ms, + grouping_labels: KeyByLabelNames::empty(), + spatial_filter_normalized: String::new(), + topk_count_events: None, + }, + query_strings: vec!["test_query".into()], + query_frequency_hz: 1.0 / 60.0, + min_t_repeat_secs: min_t, + t_repeat_gcd_secs: min_t, + } + } + + #[test] + fn always_includes_exact_fallback() { + let aqe = make_aqe(Statistic::Sum, Some(300_000), 60); + let candidates = enumerate_candidates(&aqe, 15); + assert!(candidates + .iter() + .any(|c| c.config.is_none() && c.query_method == QueryMethod::Exact)); + } + + #[test] + fn spatial_only_produces_neither_candidates() { + let aqe = make_aqe(Statistic::Sum, None, 60); + let candidates = enumerate_candidates(&aqe, 15); + // All non-EXACT candidates should be Direct (no temporal range to cover). + for c in candidates.iter().filter(|c| c.config.is_some()) { + assert_eq!(c.query_method, QueryMethod::Direct); + } + } + + #[test] + fn tumbling_w_equals_range_produces_neither() { + // range_a = 60s, scrape = 60s → only W=60, n=1 → Direct + let aqe = make_aqe(Statistic::Sum, Some(60_000), 60); + let candidates = enumerate_candidates(&aqe, 60); + for c in candidates.iter().filter(|c| c.config.is_some()) { + assert_eq!(c.query_method, QueryMethod::Direct); + } + } + + #[test] + fn mergeable_sketch_with_multiple_windows_produces_merge() { + // Min → MinMax (mergeable, not subtractable): range_a=300s, scrape=60s, min_t=300s + // → W=60 → n=5, Merge{5}. (Sum would prefer Subtract since it's also subtractable.) + let aqe = make_aqe(Statistic::Min, Some(300_000), 300); + let candidates = enumerate_candidates(&aqe, 60); + let merge_candidates: Vec<_> = candidates + .iter() + .filter(|c| matches!(c.query_method, QueryMethod::Merge { .. })) + .collect(); + assert!( + !merge_candidates.is_empty(), + "expected at least one Merge candidate" + ); + } + + #[test] + fn cms_with_heap_only_neither_no_merge() { + // CMS+Heap is neither mergeable nor subtractable → only n=1 (Direct) valid. + let aqe = make_aqe(Statistic::Topk, Some(300_000), 300); + let candidates = enumerate_candidates(&aqe, 60); + for c in candidates.iter().filter(|c| c.config.is_some()) { + assert_eq!( + c.query_method, + QueryMethod::Direct, + "CMS+Heap should only produce Direct candidates" + ); + } + } +} diff --git a/asap-planner-rs/src/optimizer/constants.rs b/asap-planner-rs/src/optimizer/constants.rs new file mode 100644 index 00000000..21c71f84 --- /dev/null +++ b/asap-planner-rs/src/optimizer/constants.rs @@ -0,0 +1,35 @@ +//! Tunable constants for candidate generation and cost modeling, centralized +//! so they're easy to find and swap for calibrated/profiled values later. + +// ponytail: small representative grids; replace with sketch-bench sweep results in Phase 3. +pub const CMS_DEPTHS: &[u64] = &[3, 5]; +pub const CMS_WIDTHS: &[u64] = &[512, 1024, 2048]; +pub const CMS_HEAP_SIZES: &[u64] = &[40, 200, 1000]; +pub const KLL_KS: &[u64] = &[200, 500]; +pub const HYDRA_ROWS: &[u64] = &[3, 5]; +pub const HYDRA_COLS: &[u64] = &[512, 1024]; +pub const HYDRA_K: u64 = 20; +pub const HLL_PRECISIONS: &[u64] = &[12, 14]; + +// Per-operation costs for one sketch instance (`AtomicCosts` defaults). +// Stub values for v1 — real numbers come from sketch-bench in Phase 3. +pub const MEM_BYTES_PER_INSTANCE: f64 = 1024.0; +pub const INSERT_CPU_SECS: f64 = 1e-7; +pub const MERGE_CPU_SECS: f64 = 1e-5; +pub const SUBTRACT_CPU_SECS: f64 = 1e-6; +pub const QUERY_CPU_SECS: f64 = 1e-5; +/// Cost of one raw/exact query execution (the EXACT_a fallback's QueryCost). +/// Without this, EXACT always wins since its IngestCost and QueryCost would +/// otherwise both be zero. +pub const EXACT_QUERY_CPU_SECS: f64 = 1e-3; + +// Global objective weights (w1..w4 in the design doc), `CostWeights` defaults. +// Real calibration (from actual cloud $/byte-sec and $/cpu-sec) is punted +// post-v1; defaults reflect that RAM-held-over-time is several orders of +// magnitude cheaper per unit than CPU-time (e.g. ~$5/GB-month vs +// ~$0.04/vCPU-hour is roughly a 1e6 ratio), so memory weights are scaled +// down accordingly rather than left equal to CPU weights. +pub const INGEST_MEM_WEIGHT: f64 = 1e-9; +pub const INGEST_CPU_WEIGHT: f64 = 1.0; +pub const QUERY_MEM_WEIGHT: f64 = 1e-9; +pub const QUERY_CPU_WEIGHT: f64 = 1.0; diff --git a/asap-planner-rs/src/optimizer/cost_model.rs b/asap-planner-rs/src/optimizer/cost_model.rs new file mode 100644 index 00000000..27c25add --- /dev/null +++ b/asap-planner-rs/src/optimizer/cost_model.rs @@ -0,0 +1,226 @@ +use asap_types::enums::WindowType; + +use super::candidate_gen::CandidateConfig; +use super::constants::{ + EXACT_QUERY_CPU_SECS, INGEST_CPU_WEIGHT, INGEST_MEM_WEIGHT, INSERT_CPU_SECS, + MEM_BYTES_PER_INSTANCE, MERGE_CPU_SECS, QUERY_CPU_SECS, QUERY_CPU_WEIGHT, QUERY_MEM_WEIGHT, + SUBTRACT_CPU_SECS, +}; +use super::sketch_properties::sketch_properties; +use super::solution::{QueryMethod, AQE}; + +/// Per-operation costs for one sketch instance. Stub defaults for v1 — real +/// values come from sketch-bench in Phase 3 (see implementation plan, 3c). +#[derive(Debug, Clone, Copy)] +pub struct AtomicCosts { + pub mem_bytes_per_instance: f64, + pub insert_cpu_secs: f64, + pub merge_cpu_secs: f64, + pub subtract_cpu_secs: f64, + pub query_cpu_secs: f64, + /// Cost of one raw/exact query execution (the EXACT_a fallback's QueryCost). + /// Without this, EXACT always wins since its IngestCost and QueryCost would + /// otherwise both be zero. + pub exact_query_cpu_secs: f64, +} + +impl Default for AtomicCosts { + fn default() -> Self { + Self { + mem_bytes_per_instance: MEM_BYTES_PER_INSTANCE, + insert_cpu_secs: INSERT_CPU_SECS, + merge_cpu_secs: MERGE_CPU_SECS, + subtract_cpu_secs: SUBTRACT_CPU_SECS, + query_cpu_secs: QUERY_CPU_SECS, + exact_query_cpu_secs: EXACT_QUERY_CPU_SECS, + } + } +} + +/// Global objective weights (w1..w4 in the design doc). Real calibration (from +/// actual cloud $/byte-sec and $/cpu-sec) is punted post-v1; defaults below +/// just reflect that RAM-held-over-time is several orders of magnitude +/// cheaper per unit than CPU-time (e.g. ~$5/GB-month vs ~$0.04/vCPU-hour is +/// roughly a 1e6 ratio), so memory weights are scaled down accordingly rather +/// than left equal to CPU weights. +#[derive(Debug, Clone, Copy)] +pub struct CostWeights { + pub ingest_mem: f64, + pub ingest_cpu: f64, + pub query_mem: f64, + pub query_cpu: f64, +} + +impl Default for CostWeights { + fn default() -> Self { + Self { + ingest_mem: INGEST_MEM_WEIGHT, + ingest_cpu: INGEST_CPU_WEIGHT, + query_mem: QUERY_MEM_WEIGHT, + query_cpu: QUERY_CPU_WEIGHT, + } + } +} + +/// IngestCost(g): steady-state cost rate of keeping `candidate` deployed, +/// independent of which AQEs query it (facility-location requirement). +/// `rho_g` is the arrival rate (items/sec) for this config's metric+filter. +pub fn ingest_cost( + candidate: &CandidateConfig, + rho_g: f64, + costs: &AtomicCosts, + weights: &CostWeights, +) -> f64 { + let Some(g) = &candidate.config else { + return 0.0; // EXACT: no streaming config deployed. + }; + + // N(s,g) = 1 if subpopulation_aware else N_g (distinct label-group count). + // N_g isn't profiled yet (needs Prometheus series-count data) — use 1 as a + // placeholder; both branches collapse to the same value until that lands. + let n = 1.0_f64; + + // Defensive floor: slide_interval is a plain u64 on a widely-shared struct; + // guard against div-by-zero producing `inf` and poisoning cost comparisons. + let n_concurrent = match g.window_type { + WindowType::Tumbling => 1.0, + WindowType::Sliding => (g.window_size as f64 / g.slide_interval.max(1) as f64).ceil(), + }; + + let mem_active = n_concurrent * n * costs.mem_bytes_per_instance; + let mem_retain = match g.window_type { + WindowType::Tumbling => candidate.n_windows as f64 * n * costs.mem_bytes_per_instance, + WindowType::Sliding => 0.0, // already counted in mem_active's concurrent windows + }; + + let cpu_ingest = match g.window_type { + WindowType::Tumbling => rho_g * costs.insert_cpu_secs, + WindowType::Sliding => rho_g * n_concurrent * costs.insert_cpu_secs, + }; + + weights.ingest_mem * (mem_active + mem_retain) + weights.ingest_cpu * cpu_ingest +} + +/// QueryCost(a,g): cost of answering one query for `a` from `candidate`. +pub fn query_cost( + _a: &AQE, + candidate: &CandidateConfig, + costs: &AtomicCosts, + weights: &CostWeights, +) -> f64 { + let Some(g) = &candidate.config else { + return costs.exact_query_cpu_secs * weights.query_cpu; // EXACT: raw query at query time. + }; + + let n = 1.0_f64; // N(s,g); see ingest_cost comment. + let props = sketch_properties(g.aggregation_type); + + let (cpu, mem) = match &candidate.query_method { + QueryMethod::Direct => (n * costs.query_cpu_secs, n * costs.mem_bytes_per_instance), + QueryMethod::Merge { num_windows } => { + debug_assert!(props.mergeable); + let merges = (*num_windows).saturating_sub(1) as f64; + ( + n * (merges * costs.merge_cpu_secs + costs.query_cpu_secs), + *num_windows as f64 * n * costs.mem_bytes_per_instance, + ) + } + QueryMethod::Subtract => { + debug_assert!(props.subtractable); + ( + n * (costs.subtract_cpu_secs + costs.query_cpu_secs), + 2.0 * n * costs.mem_bytes_per_instance, + ) + } + // candidate_gen only ever pairs Exact with config=None, already handled above. + QueryMethod::Exact => { + unreachable!("Exact query_method must not be paired with Some(config)") + } + }; + + weights.query_cpu * cpu + weights.query_mem * mem +} + +/// Total cost rate contributed by assigning AQE `a` (with frequency `f_a` = +/// `a.query_frequency_hz`) to `candidate`: IngestCost(g) + f_a * QueryCost(a,g). +/// This is the per-(a,g) term the greedy/MIP solver minimizes. +pub fn total_cost_rate( + a: &AQE, + candidate: &CandidateConfig, + rho_g: f64, + costs: &AtomicCosts, + weights: &CostWeights, +) -> f64 { + ingest_cost(candidate, rho_g, costs, weights) + + a.query_frequency_hz * query_cost(a, candidate, costs, weights) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::optimizer::candidate_gen::enumerate_candidates; + use asap_types::query_requirements::QueryRequirements; + use promql_utilities::data_model::KeyByLabelNames; + use promql_utilities::query_logics::enums::Statistic; + + fn make_aqe(stat: Statistic, range_ms: Option, min_t: u64) -> AQE { + AQE { + requirements: QueryRequirements { + metric: "test_metric".into(), + statistics: vec![stat], + data_range_ms: range_ms, + grouping_labels: KeyByLabelNames::empty(), + spatial_filter_normalized: String::new(), + topk_count_events: None, + }, + query_strings: vec!["test_query".into()], + query_frequency_hz: 1.0 / 60.0, + min_t_repeat_secs: min_t, + t_repeat_gcd_secs: min_t, + } + } + + #[test] + fn exact_has_zero_ingest_cost_and_nonzero_query_cost() { + let candidate = CandidateConfig { + config: None, + query_method: QueryMethod::Exact, + n_windows: 0, + }; + let costs = AtomicCosts::default(); + let weights = CostWeights::default(); + let a = make_aqe(Statistic::Sum, Some(300_000), 300); + assert_eq!(ingest_cost(&candidate, 1.0, &costs, &weights), 0.0); + assert!(query_cost(&a, &candidate, &costs, &weights) > 0.0); + } + + #[test] + fn subtract_is_cheaper_than_merge_for_the_same_window_count() { + // Calibration-independent: Subtract is O(1) (one subtract + one read) while + // Merge is O(n) (n-1 merges + one read), so for the same n and same + // underlying config, Subtract must cost less regardless of weight tuning. + let a = make_aqe(Statistic::Sum, Some(300_000), 300); + let candidates = enumerate_candidates(&a, 60); + let template = candidates + .iter() + .find_map(|c| c.config.clone()) + .expect("expected at least one deployed candidate"); + + let costs = AtomicCosts::default(); + let weights = CostWeights::default(); + let merge = CandidateConfig { + config: Some(template.clone()), + query_method: QueryMethod::Merge { num_windows: 5 }, + n_windows: 5, + }; + let subtract = CandidateConfig { + config: Some(template), + query_method: QueryMethod::Subtract, + n_windows: 5, + }; + + assert!( + query_cost(&a, &subtract, &costs, &weights) < query_cost(&a, &merge, &costs, &weights) + ); + } +} diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs new file mode 100644 index 00000000..ffb0fe59 --- /dev/null +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -0,0 +1,164 @@ +use std::collections::HashMap; + +use asap_types::aggregation_config::AggregationConfig; +use tracing::debug; + +use super::candidate_gen::enumerate_candidates; +use super::cost_model::{ingest_cost, query_cost, total_cost_rate, AtomicCosts, CostWeights}; +use super::solution::{AQEAssignment, OptimizerSolution, AQE}; + +/// Greedily assign each AQE to its independently-cheapest candidate config. +/// +/// No cross-AQE sharing: every deployed sketch serves exactly one AQE, even if +/// two AQEs could share one. The Phase 3 MIP finds sharing opportunities; this +/// is the v1 baseline. +/// +/// `rho_g` is the per-item arrival rate used for every candidate's IngestCost. +/// Real per-config rates need Prometheus scrape-rate × series-count data, +/// which isn't wired up yet — a single placeholder value is applied uniformly. +pub fn greedy_assign( + aqes: Vec, + scrape_interval_secs: u64, + rho_g: f64, + costs: &AtomicCosts, + weights: &CostWeights, +) -> OptimizerSolution { + let mut deployed_configs: HashMap = HashMap::new(); + let mut assignments = Vec::new(); + let mut estimated_ingest_cost_per_sec = 0.0; + let mut estimated_total_cost_per_sec = 0.0; + let mut next_id: u64 = 1; + + for aqe in aqes { + let candidates = enumerate_candidates(&aqe, scrape_interval_secs); + + let best = candidates + .into_iter() + .map(|c| { + let cost = total_cost_rate(&aqe, &c, rho_g, costs, weights); + (c, cost) + }) + // total_cmp (not partial_cmp().unwrap()) so a stray NaN cost can't panic. + .min_by(|(_, a), (_, b)| a.total_cmp(b)) + .map(|(c, _)| c) + .expect("enumerate_candidates always returns at least the EXACT fallback"); + + let ingest = ingest_cost(&best, rho_g, costs, weights); + let query_rate = aqe.query_frequency_hz * query_cost(&aqe, &best, costs, weights); + let query_method = best.query_method.clone(); + + let aggregation_id = match best.config { + None => None, + Some(mut config) => { + let id = next_id; + next_id += 1; + config.aggregation_id = id; + deployed_configs.insert(id, config); + Some(id) + } + }; + + debug!( + metric = %aqe.requirements.metric, + aggregation_id = ?aggregation_id, + query_method = ?query_method, + ingest_cost_per_sec = ingest, + query_cost_per_sec = query_rate, + "greedy: assigned AQE" + ); + + estimated_ingest_cost_per_sec += ingest; + estimated_total_cost_per_sec += ingest + query_rate; + + assignments.push(AQEAssignment { + aqe, + aggregation_id, + query_method, + estimated_query_cost_per_sec: query_rate, + }); + } + + OptimizerSolution { + deployed_configs, + assignments, + estimated_ingest_cost_per_sec, + estimated_total_cost_per_sec, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::query_requirements::QueryRequirements; + use promql_utilities::data_model::KeyByLabelNames; + use promql_utilities::query_logics::enums::Statistic; + use std::collections::HashMap as StdHashMap; + + fn make_aqe(stat: Statistic, range_ms: Option, min_t: u64, freq_hz: f64) -> AQE { + AQE { + requirements: QueryRequirements { + metric: "test_metric".into(), + statistics: vec![stat], + data_range_ms: range_ms, + grouping_labels: KeyByLabelNames::empty(), + spatial_filter_normalized: String::new(), + topk_count_events: None, + }, + query_strings: vec!["test_query".into()], + query_frequency_hz: freq_hz, + min_t_repeat_secs: min_t, + t_repeat_gcd_secs: min_t, + } + } + + #[test] + fn assigns_unique_ids_to_each_deployed_config() { + let aqes = vec![ + make_aqe(Statistic::Min, Some(300_000), 300, 1.0 / 60.0), + make_aqe(Statistic::Max, Some(300_000), 300, 1.0 / 60.0), + ]; + let solution = greedy_assign( + aqes, + 60, + 1.0, + &AtomicCosts::default(), + &CostWeights::default(), + ); + + let mut seen_ids: StdHashMap = StdHashMap::new(); + for (id, _) in solution.deployed_configs.iter() { + assert!( + seen_ids.insert(*id, ()).is_none(), + "duplicate aggregation_id" + ); + } + assert_eq!(solution.assignments.len(), 2); + } + + #[test] + fn unsupported_multi_statistic_aqe_falls_back_to_exact() { + let aqe = AQE { + requirements: QueryRequirements { + metric: "test_metric".into(), + statistics: vec![Statistic::Sum, Statistic::Count], // avg-style, unsupported + data_range_ms: Some(60_000), + grouping_labels: KeyByLabelNames::empty(), + spatial_filter_normalized: String::new(), + topk_count_events: None, + }, + query_strings: vec!["avg_query".into()], + query_frequency_hz: 1.0 / 60.0, + min_t_repeat_secs: 60, + t_repeat_gcd_secs: 60, + }; + let solution = greedy_assign( + vec![aqe], + 60, + 1.0, + &AtomicCosts::default(), + &CostWeights::default(), + ); + assert_eq!(solution.num_exact_fallback(), 1); + assert!(solution.deployed_configs.is_empty()); + } +} diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs new file mode 100644 index 00000000..534f4f60 --- /dev/null +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -0,0 +1,18 @@ +pub mod aqe_extractor; +pub mod candidate_gen; +pub mod constants; +pub mod cost_model; +pub mod greedy; +pub mod pipeline; +pub mod sketch_properties; +pub mod solution; +pub mod translator; + +pub use aqe_extractor::{extract_aqes, RQE}; +pub use candidate_gen::{enumerate_candidates, CandidateConfig}; +pub use cost_model::{ingest_cost, query_cost, total_cost_rate, AtomicCosts, CostWeights}; +pub use greedy::greedy_assign; +pub use pipeline::{run_all_exact_pipeline, run_greedy_pipeline}; +pub use sketch_properties::{sketch_properties, SketchProperties}; +pub use solution::{AQEAssignment, OptimizerSolution, QueryMethod, AQE}; +pub use translator::translate; diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs new file mode 100644 index 00000000..e0132cfa --- /dev/null +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -0,0 +1,148 @@ +use asap_types::inference_config::InferenceConfig; +use asap_types::streaming_config::StreamingConfig; +use asap_types::PromQLSchema; + +use crate::config::input::ControllerConfig; + +use super::aqe_extractor::{extract_aqes, RQE}; +use super::cost_model::{AtomicCosts, CostWeights}; +use super::greedy::greedy_assign; +use super::solution::OptimizerSolution; +use super::translator::{translate, TranslationSummary}; + +/// Run the all-EXACT optimizer pipeline (Phase 1 scaffolding). +/// +/// Converts a `ControllerConfig` into `(StreamingConfig, InferenceConfig)` via +/// the optimizer path: RQEs → AQEs → all-EXACT solution → deployment artifacts. +/// +/// No streaming configs are deployed — every AQE falls back to raw data at +/// query time. This validates the end-to-end pipeline plumbing before real +/// sketch selection logic is added in Phase 2. +pub fn run_all_exact_pipeline( + config: &ControllerConfig, + schema: &PromQLSchema, +) -> (StreamingConfig, InferenceConfig) { + let rqes = config_to_rqes(config); + let aqes = extract_aqes(&rqes, schema); + let solution = OptimizerSolution::all_exact(aqes); + + let summary = TranslationSummary::from_solution(&solution); + tracing::info!( + num_deployed_configs = summary.num_deployed_configs, + num_sketch_assignments = summary.num_sketch_assignments, + num_exact_fallbacks = summary.num_exact_fallbacks, + "optimizer pipeline: all-EXACT solution produced" + ); + + translate(&solution) +} + +/// Run the greedy optimizer pipeline (Phase 2): each AQE is assigned, independently, +/// to its cheapest feasible candidate config (or to the EXACT fallback). +/// +/// No cross-AQE sharing — every deployed sketch serves exactly one AQE, even +/// if two AQEs could share one. The Phase 3 MIP finds sharing opportunities. +/// +/// `rho_g` is a placeholder arrival rate applied uniformly to every candidate's +/// IngestCost; real per-config rates need Prometheus scrape-rate × series-count +/// data, which isn't wired up yet (see implementation plan TODOs). +pub fn run_greedy_pipeline( + config: &ControllerConfig, + schema: &PromQLSchema, + scrape_interval_secs: u64, + rho_g: f64, +) -> (StreamingConfig, InferenceConfig) { + let rqes = config_to_rqes(config); + let aqes = extract_aqes(&rqes, schema); + let solution = greedy_assign( + aqes, + scrape_interval_secs, + rho_g, + &AtomicCosts::default(), + &CostWeights::default(), + ); + + let summary = TranslationSummary::from_solution(&solution); + tracing::info!( + num_deployed_configs = summary.num_deployed_configs, + num_sketch_assignments = summary.num_sketch_assignments, + num_exact_fallbacks = summary.num_exact_fallbacks, + estimated_ingest_cost_per_sec = solution.estimated_ingest_cost_per_sec, + estimated_total_cost_per_sec = solution.estimated_total_cost_per_sec, + "optimizer pipeline: greedy solution produced" + ); + + translate(&solution) +} + +/// Convert a `ControllerConfig`'s query groups into a flat list of RQEs. +/// Each (query, repetition_delay) pair becomes one RQE. +fn config_to_rqes(config: &ControllerConfig) -> Vec { + config + .query_groups + .iter() + .flat_map(|qg| { + qg.queries.iter().map(|q| RQE { + query_string: q.clone(), + t_repeat_secs: qg.repetition_delay, + }) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_config(queries: &[(&str, u64)]) -> ControllerConfig { + use crate::config::input::QueryGroup; + + let query_groups = queries + .iter() + .map(|(q, t)| QueryGroup { + id: None, + queries: vec![q.to_string()], + repetition_delay: *t, + controller_options: Default::default(), + step: None, + range_duration: None, + }) + .collect(); + + ControllerConfig { + query_groups, + sketch_parameters: None, + aggregate_cleanup: None, + metrics: None, + existing_streaming_config: None, + existing_inference_config: None, + } + } + + #[test] + fn all_exact_pipeline_produces_empty_streaming_config() { + let config = make_config(&[("sum_over_time(metric[5m])", 60), ("sum(other_metric)", 30)]); + let schema = PromQLSchema::new(); + let (streaming, _inference) = run_all_exact_pipeline(&config, &schema); + // All-EXACT: no streaming configs deployed. + assert!(streaming.get_all_aggregation_configs().is_empty()); + } + + #[test] + fn greedy_pipeline_deploys_a_config_for_a_mergeable_aqe() { + let config = make_config(&[("min_over_time(metric[5m])", 60)]); + let schema = PromQLSchema::new(); + let (streaming, inference) = run_greedy_pipeline(&config, &schema, 60, 1.0); + assert!(!streaming.get_all_aggregation_configs().is_empty()); + assert!(!inference.query_configs.is_empty()); + } + + #[test] + fn config_to_rqes_flattens_groups() { + let config = make_config(&[("sum_over_time(a[5m])", 60), ("sum_over_time(b[5m])", 30)]); + let rqes = config_to_rqes(&config); + assert_eq!(rqes.len(), 2); + assert_eq!(rqes[0].t_repeat_secs, 60); + assert_eq!(rqes[1].t_repeat_secs, 30); + } +} diff --git a/asap-planner-rs/src/optimizer/sketch_properties.rs b/asap-planner-rs/src/optimizer/sketch_properties.rs new file mode 100644 index 00000000..2f0e9189 --- /dev/null +++ b/asap-planner-rs/src/optimizer/sketch_properties.rs @@ -0,0 +1,70 @@ +use promql_utilities::query_logics::enums::AggregationType; + +#[derive(Debug, Clone, Copy)] +pub struct SketchProperties { + /// Two instances can be combined into one representing the union. + pub mergeable: bool, + /// Element-wise difference is defined; enables the Subtract query method (tumbling only). + pub subtractable: bool, + /// One deployed instance handles multiple label-group keys simultaneously. + pub subpopulation_aware: bool, +} + +pub fn sketch_properties(t: AggregationType) -> SketchProperties { + let p = |me, su, sp| SketchProperties { + mergeable: me, + subtractable: su, + subpopulation_aware: sp, + }; + match t { + AggregationType::Sum => p(true, true, false), + AggregationType::Increase => p(true, false, false), + AggregationType::MinMax => p(true, false, false), + AggregationType::DatasketchesKLL => p(true, false, false), + AggregationType::MultipleSum => p(true, true, true), + AggregationType::MultipleIncrease => p(true, false, true), + AggregationType::MultipleMinMax => p(true, false, true), + AggregationType::HydraKLL => p(true, false, true), + AggregationType::CountMinSketch => p(true, true, true), + // ponytail: heap top-k lists don't compose across windows; CMS cells do but the + // combined type requires the heap, so neither merging nor subtracting is safe here. + AggregationType::CountMinSketchWithHeap => p(false, false, true), + AggregationType::SetAggregator | AggregationType::DeltaSetAggregator => { + p(true, false, false) + } + AggregationType::HLL => p(true, false, false), + // Legacy wrapper types: properties unknown; treat conservatively. + AggregationType::SingleSubpopulation | AggregationType::MultipleSubpopulation => { + p(false, false, false) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cms_is_mergeable_subtractable_subpop_aware() { + let p = sketch_properties(AggregationType::CountMinSketch); + assert!(p.mergeable && p.subtractable && p.subpopulation_aware); + } + + #[test] + fn cms_with_heap_not_mergeable_not_subtractable() { + let p = sketch_properties(AggregationType::CountMinSketchWithHeap); + assert!(!p.mergeable && !p.subtractable && p.subpopulation_aware); + } + + #[test] + fn sum_mergeable_subtractable_not_subpop() { + let p = sketch_properties(AggregationType::Sum); + assert!(p.mergeable && p.subtractable && !p.subpopulation_aware); + } + + #[test] + fn kll_mergeable_not_subtractable() { + let p = sketch_properties(AggregationType::DatasketchesKLL); + assert!(p.mergeable && !p.subtractable); + } +} diff --git a/asap-planner-rs/src/optimizer/solution.rs b/asap-planner-rs/src/optimizer/solution.rs new file mode 100644 index 00000000..5e0924c3 --- /dev/null +++ b/asap-planner-rs/src/optimizer/solution.rs @@ -0,0 +1,138 @@ +use std::collections::HashMap; + +use asap_types::aggregation_config::AggregationConfig; +use asap_types::query_requirements::QueryRequirements; + +/// An atomic query expression: one leaf aggregation extracted from a QE tree, +/// together with the optimizer-level metadata needed to assign it a config. +#[derive(Debug, Clone)] +pub struct AQE { + /// What the query needs (metric, statistics, range, labels, spatial filter). + pub requirements: QueryRequirements, + + /// Original PromQL/SQL query strings from all RQEs that contain this AQE. + /// Preserved for use by the translator when building InferenceConfig. + pub query_strings: Vec, + + /// Query frequency in Hz: f_a (this field) = Σ_{r ∈ R_a} 1/T_r. + /// Used in the MIP objective to convert per-query QueryCost into a cost + /// rate (cost/sec) commensurate with the continuously-accruing IngestCost. + /// Represents the total query load from all dashboards independently + /// hitting the sketch. + pub query_frequency_hz: f64, + + /// Minimum repeat interval across all RQEs that reference this AQE (secs). + /// Determines the freshness constraint on the window size: W ≤ min_t_repeat + /// ensures a completed window is available for every dashboard's cycle. + /// When multiple RQEs share this AQE, the fastest dashboard is the binding + /// constraint. + pub min_t_repeat_secs: u64, + + /// GCD of all repeat intervals across RQEs that reference this AQE (secs). + /// The natural candidate for the slide interval S: windows that complete + /// every GCD seconds align harmonically with all dashboard refresh cycles, + /// ensuring every dashboard can always be served a fresh result on-cycle. + pub t_repeat_gcd_secs: u64, +} + +/// How an AQE is answered from its assigned streaming config. +/// +/// Determined by (ingest_type, W vs range_a, sketch algebra) — not a free +/// decision variable. See the compatibility table in the design doc. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum QueryMethod { + /// W = range_a: one completed window covers the query range exactly. + /// Direct read, no merge or subtract needed. + Direct, + + /// W < range_a, sketch is mergeable: combine `num_windows` retained + /// tumbling sub-windows at query time. Cost scales linearly with num_windows. + Merge { num_windows: u64 }, + + /// W < range_a, sketch is subtractable: subtract two prefix-sum checkpoints. + /// O(1) cost regardless of range_a/W. + Subtract, + + /// No streaming config deployed for this AQE — query raw/exact data at + /// query time. Corresponds to the EXACT_a fallback (IngestCost = 0, Error = 0). + Exact, +} + +/// The assignment of a single AQE to a streaming config (or EXACT fallback). +#[derive(Debug, Clone)] +pub struct AQEAssignment { + pub aqe: AQE, + + /// ID of the deployed config that serves this AQE. + /// `None` means the EXACT_a fallback (no streaming config, raw query). + pub aggregation_id: Option, + + /// How this AQE's answer is derived from the assigned config. + pub query_method: QueryMethod, + + /// Estimated cost rate for this assignment: f_a * QueryCost(a, g), where + /// f_a is `aqe.query_frequency_hz`. + /// Zero for Exact assignments (IngestCost is also zero). + pub estimated_query_cost_per_sec: f64, +} + +/// The output of the optimizer: a complete plan for a given RQE workload. +/// +/// Contains the set of streaming configs to deploy and the assignment of every +/// AQE to one of those configs (or to the EXACT fallback). A thin translator +/// converts this into `StreamingConfig + InferenceConfig` deployment artifacts. +#[derive(Debug, Clone)] +pub struct OptimizerSolution { + /// Deployed streaming configs (y_g = 1 in the MIP). Keyed by aggregation_id. + /// Empty for all-EXACT solutions (Phase 1 scaffolding). + pub deployed_configs: HashMap, + + /// One entry per deduplicated AQE across the full RQE workload. + pub assignments: Vec, + + /// Estimated steady-state ingestion cost rate across all deployed configs + /// (Σ_{g: y_g=1} IngestCost(g)). + pub estimated_ingest_cost_per_sec: f64, + + /// Estimated total cost rate: ingest + query components combined. + pub estimated_total_cost_per_sec: f64, +} + +impl OptimizerSolution { + /// Construct an all-EXACT solution: every AQE falls back to raw data, + /// no streaming configs are deployed. Used as the Phase 1 scaffolding baseline. + pub fn all_exact(aqes: Vec) -> Self { + let assignments = aqes + .into_iter() + .map(|aqe| AQEAssignment { + aqe, + aggregation_id: None, + query_method: QueryMethod::Exact, + estimated_query_cost_per_sec: 0.0, + }) + .collect(); + + Self { + deployed_configs: HashMap::new(), + assignments, + estimated_ingest_cost_per_sec: 0.0, + estimated_total_cost_per_sec: 0.0, + } + } + + /// Number of AQEs served by an approximate sketch (not EXACT fallback). + pub fn num_sketch_served(&self) -> usize { + self.assignments + .iter() + .filter(|a| a.query_method != QueryMethod::Exact) + .count() + } + + /// Number of AQEs falling back to exact/raw computation. + pub fn num_exact_fallback(&self) -> usize { + self.assignments + .iter() + .filter(|a| a.query_method == QueryMethod::Exact) + .count() + } +} diff --git a/asap-planner-rs/src/optimizer/translator.rs b/asap-planner-rs/src/optimizer/translator.rs new file mode 100644 index 00000000..b96ae4fe --- /dev/null +++ b/asap-planner-rs/src/optimizer/translator.rs @@ -0,0 +1,81 @@ +use asap_types::aggregation_reference::AggregationReference; +use asap_types::inference_config::InferenceConfig; +use asap_types::query_config::QueryConfig; +use asap_types::streaming_config::StreamingConfig; + +use super::solution::{OptimizerSolution, QueryMethod}; + +/// Translate an `OptimizerSolution` into the deployment artifacts consumed by +/// Arroyo and the query engine. +/// +/// Phase 1 (all-EXACT): deployed_configs is empty, all assignments are Exact, +/// so both output structs are empty/stub. Real translation logic fills in as +/// Phase 2/3 add sketch configs to the solution. +pub fn translate(solution: &OptimizerSolution) -> (StreamingConfig, InferenceConfig) { + let streaming_config = build_streaming_config(solution); + let inference_config = build_inference_config(solution); + (streaming_config, inference_config) +} + +fn build_streaming_config(solution: &OptimizerSolution) -> StreamingConfig { + // Deployed configs map directly to AggregationConfigs — the types are the same. + StreamingConfig::new(solution.deployed_configs.clone()) +} + +fn build_inference_config(solution: &OptimizerSolution) -> InferenceConfig { + use asap_types::enums::{CleanupPolicy, QueryLanguage}; + + let mut inference = InferenceConfig::new(QueryLanguage::promql, CleanupPolicy::NoCleanup); + + // For Phase 1 (all-EXACT), every assignment has aggregation_id = None, so + // this loop emits nothing — the inference engine falls back to raw + // querying for all AQEs, matching the all-EXACT solution. + for assignment in &solution.assignments { + let Some(aggregation_id) = assignment.aggregation_id else { + continue; + }; + let retain = retention_count_for_assignment(&assignment.query_method); + let agg_ref = AggregationReference::new(aggregation_id, Some(retain)); + + for query_string in &assignment.aqe.query_strings { + inference + .query_configs + .push(QueryConfig::new(query_string.clone()).add_aggregation(agg_ref.clone())); + } + } + + inference +} + +/// For a Merge assignment, the number of retained windows to configure in the +/// inference config (num_aggregates_to_retain on the AggregationReference). +pub fn retention_count_for_assignment(query_method: &QueryMethod) -> u64 { + match query_method { + QueryMethod::Direct => 1, + QueryMethod::Merge { num_windows } => *num_windows, + // Subtract needs 2 prefix checkpoints per query but we retain + // ceil(range_a/W) checkpoints total to cover the full lookback. + // The exact value comes from the config's window parameters; use 1 + // as a placeholder until Phase 3 wires this up properly. + QueryMethod::Subtract => 1, + QueryMethod::Exact => 0, + } +} + +/// Summary of what the translator produced, for logging/debugging. +#[derive(Debug)] +pub struct TranslationSummary { + pub num_deployed_configs: usize, + pub num_sketch_assignments: usize, + pub num_exact_fallbacks: usize, +} + +impl TranslationSummary { + pub fn from_solution(solution: &OptimizerSolution) -> Self { + Self { + num_deployed_configs: solution.deployed_configs.len(), + num_sketch_assignments: solution.num_sketch_served(), + num_exact_fallbacks: solution.num_exact_fallback(), + } + } +} diff --git a/asap-planner-rs/src/promql/generator.rs b/asap-planner-rs/src/promql/generator.rs index 486d9c3b..63761492 100644 --- a/asap-planner-rs/src/promql/generator.rs +++ b/asap-planner-rs/src/promql/generator.rs @@ -18,7 +18,12 @@ use crate::RuntimeOptions; /// `(query_string, Vec<(identifying_key, cleanup_param)>)` pairs produced by binary leaf decomposition. type LeafEntries = Vec<(String, Vec<(String, Option)>)>; -/// Run the full planning pipeline and produce YAML outputs +/// Run the full planning pipeline and produce YAML outputs. +/// +/// This is the hardcoded sketch/window selection path that `Controller::generate()` +/// currently calls. `crate::optimizer` implements an optimization-based replacement +/// (issue #405) but is not wired in here yet — see `bin/optimizer_cli.rs` for an +/// offline runner and `.design_docs/optimizer-v1-implementation-plan.md` for status. pub fn generate_plan( controller_config: &ControllerConfig, schema: &PromQLSchema,