Repository navigation
feat(data-plane): read query-time raw windows from a local raw buffer and Prometheus remote read - #831
Draft
zzylol wants to merge 16 commits into
Draft
feat(data-plane): read query-time raw windows from a local raw buffer and Prometheus remote read#831zzylol wants to merge 16 commits into
zzylol wants to merge 16 commits into
Conversation
…ecution stack The backend now executes ASAPPlanner's selected PhysicalASAPDAG as given (#509): the Planner runs Stages 0-3, the backend supplies deployment inputs. - Pin every Planner crate the backend uses to ASAPPlanner main (58999df2): asap-types, asap-frontend-promql, asap-logical-asap-optimization, asap-plan-selection, asap-executor. The sync script updates all five. - control_plane: lower a PromQL PlanningWorkload, run Stages 1-3 (run_stage1_to_stage3) under DeploymentInputs (DeploymentCapabilities and PlanSelectionSettings), and write the selected candidate's DAG, with each PaneChain expanded into its panes, plus the query each root answers (plan.json). - data_plane: load and validate plan.json, accept Remote Write, forward queries to Prometheus until the query engine lands. - Remove the backend's own plan selection, placement pricing, ERP, SDS, precompute and query engines, ClickHouse, OTel and OpAMP paths, which were built on the removed Planner APIs.
…puts Each MatAtIngestion output the query side reads (the executor's materialization frontier) reads one tumbling pane of raw samples. Outputs that differ only in which pane they read share one SummaryDefinition, identified by a SHA-256 of their sub-DAG with node ids blanked and offset TimeShifts skipped. Each output binds to its definition with an offset: at evaluation time T it reads the pane ending at T - offset. SummaryStore keeps built panes by definition and end. A plan with a node kept across evaluations is rejected: the backend declares it cannot keep one. #509 Example 4b on a deployment without raw data: its six 10-minute KLL panes become one definition with offsets 0..50 min.
An engine thread owns the installed plan (Planner IR holds Rcs) and receives Remote Write samples from the HTTP server. For each stored summary definition it buffers its Scan metric's samples by tumbling pane (end - width, end], like PromQL ranges. When the watermark (newest sample minus --allowed-lateness-ms) passes a pane's end, it runs the plan's Scan over them (applying label matchers) and the definition's build from its TimeRange input, then stores the output. Panes without samples are stored empty; the first pane seen is skipped because it may be partial; samples for built panes are dropped and counted. A full engine queue answers Remote Write with 503 so senders retry.
Each plan root is compiled once with two kinds of inputs: stored outputs, bound at evaluation time T to their definition's pane ending at T - offset, and query-time raw ranges (TimeRange <- TimeShift* <- Scan), bound to the metric's samples in (T - offset - range, T - offset] fetched from Prometheus and filtered by the plan's Scan. An instant query whose text is a planned query is answered with a Prometheus vector and x-asap-execution: plan. Unplanned queries are forwarded; planned ones that cannot be answered (pane missing or off the pane grid, fetch or execution failure) are forwarded with x-asap-forward-reason.
For Examples 1, 3a, 3b, 4a and 4b: plan the workload, start the data_plane binary against a Prometheus holding generated samples, Remote Write the same samples, and require every planned query to be answered by the plan with the Planner executor's result for the selected logical candidate with nothing materialized (control_plane::plan_selected_at_query_time). 3b also runs, and 4b only runs, on a deployment without raw data, so their panes are built from Remote Write. Exact results must match exactly; Example 3a/4a's five-year KLL windows exceed k and KLL compaction is randomized, so they must agree within twice the 0.005 rank target.
A plan node materialized `MatAtQueryKeptAcrossEvaluations` (#509 Example 4's B3) is now executed instead of rejected. Its output is kept in memory by (definition, pane end), with the SDS identity rule, built from its pane's raw samples the first time an evaluation reads it, reused by later evaluations, and evicted like stored panes (older than the largest read offset plus two pane widths, measured from the newest evaluation). The control plane's default DeploymentCapabilities now set query_time_results_retained. examples/planner-layering-example4-b3.json is a workload for which the Planner selects B3 under the default cost model.
A pane computed at query time is kept across evaluations only once its end is at or before the precompute engine's watermark (newest sample minus the allowed lateness). A pane read while still open is computed for that evaluation but not kept, so a later evaluation computes it again, including samples that arrived late, and keeps it once it is closed.
A planned /api/v1/query_range runs one evaluation per step, start + k*step through end, and returns a Prometheus matrix. Each query-time raw range is fetched from Prometheus once for all steps and sliced per step; each kept definition's missing panes are fetched once, so a kept pane built at one step is reused by the later ones. Stored panes are read per step. If any step cannot be answered, the whole request is forwarded with the reason. start/end accept Unix seconds or RFC 3339 (now also for instant queries), step float seconds or a Prometheus duration; requests Prometheus would reject, and requests with a nonzero limit, are forwarded unchanged.
…and asap-executor only Re-pin ASAPPlanner to main f0dc1825 and use its handoff API. - The control plane calls `asap_planner::plan_installable` once and writes the returned `InstallablePlan` as plan.json. The backend's own pane-chain expansion, Stage 2/3 plumbing and query-time reference rebuild are gone; tests use `InstallablePlan::query_time_reference`. - The data plane reads `stored_outputs` (built_at, pane window, retention_ms) and `query_time_raw_windows`, and selects every pane's and raw window's rows with `asap_executor::RawSampleSelector`. It no longer uses `frontier_from_timing` or `promql_rows`. - The SDS keeps its own identity (sub-DAG hash, offset-only TimeShifts transparent) and keeps each definition for the largest `retention_ms` among its outputs, measured from the newest evaluation that can read it, instead of "largest offset plus two pane widths". - Among Planner crates the workspace depends on `asap-planner` and `asap-executor` only; IR types come through `asap_executor::planner`. - `--plan-selection-settings` on the control plane and `--execution-memory-limit-mib` on the data plane, which the benefit harness passes.
…a shared pane store - Ingestion: N workers (`--ingestion-workers`, default one per core). Remote Write series are sharded by a hash of their label set; each worker owns its shard's open panes. The global watermark is the minimum over the workers that have seen a sample; when it passes a pane, every worker builds its part, and the last part completes the pane: parts are joined and the states of a group several workers saw are merged with the family's merge. A request is queued to every worker or refused whole. - Queries: a pool of `--query-threads` threads, one query (and every step of a range query) per thread. The executor's compiled plans hold `Rc`s, so each thread reads plan.json and compiles its own. - SDS: `RwLock<HashMap<(definition, pane end), Arc<pane>>>`. Panes are inserted once and never rewritten; readers clone the `Arc`. Kept query-time results use the same store, shared by the query threads. A stored pane binds to a root as a shared source, copied once per run (the executor's sources emit owned batches) instead of twice. - jemalloc is the data plane's global allocator.
…he watermark it announces Two races of the sharded engine, both found by a test that writes a pane's samples and queries it at once: - A planned query ran on the query pool while the ingestion workers could still be ingesting the Remote Write requests answered before it, so the newest pane was missing and the query was forwarded. A query now waits until every worker has run what was queued before it (`sync`), as the single engine thread's queue ordered it. - A worker closed its panes at the watermark it read before another worker moved it, then announced the newer watermark to the other workers but not to itself, so its part of the newest pane never arrived. It now closes and announces the same watermark.
The ingestion side of a stored definition, Scan -> TimeShift* -> TimeRange(pane) -> SummaryAgg, runs per sample instead of over each pane's buffered rows: - per series, once: the executor's RawSampleSelector turns the series' first sample into its input row (applying the Scan's matchers), and the row's group columns give its group; cached by label set; - per sample: the open pane's state for that group is updated in place with the updater the executor's batch build uses for the family (create_planner_accumulator), with the same value and timestamp; - per group, once: the definition's compiled build runs over the group's first row, and its output row is the template each closed pane's row is made from (pane timestamp and state replaced). A build of another shape (row filter, keyed or shared-grid family, operators between the range and the summary, matchers reading the value or time) keeps the batch build. A parity test checks every family the examples store (counter rate, sum, KLL at two pane widths) against the executor's batch build of the same samples: exact states byte for byte, KLL by quantile (KLL seeds its compaction randomly per sketch).
… and Prometheus remote read - A local raw buffer keeps the recent samples Remote Write delivered, sharded like ingestion: each ingestion worker appends its shard. It keeps the plan's longest query-time raw read back from the newest sample: the largest `end_offset + width` of `query_time_raw_windows` and of the panes built at query time and kept (their retention, which covers range evaluations). `--raw-buffer-ms` overrides it; 0 keeps none. - A raw window is read from the buffer; the part before the buffer's cut (its first sample, or the trimmed horizon of any shard) is read with Prometheus remote read (protobuf, `/api/v1/read`). The JSON query API is no longer used for raw samples. Forwarding of unplanned queries is unchanged. - The data plane logs the buffer's series and samples every 30 s.
…est query-time read The buffer kept every metric for the plan's single longest raw read: in the benefit plan, 60 minutes of the counter whose raw reads span one minute, and of metrics no raw window reads. It now keeps only the metrics the plan reads at query time, each for its own longest read (query-time raw windows, and panes built at query time and kept).
zzylol
added
enhancement
zzylol
force-pushed
the
feat/per-sample-ingestion
branch
2 times, most recently
from
October 21, 2026 00:21
d68efef to
41ccfbe
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stack 4/4. Base: #830.
Why
Every query-time raw window, and every missing kept pane, was an instant range-selector query to Prometheus decoded as JSON. With raw data retained that was 15.5% of the data plane's CPU (146 MB of JSON per 90 minutes at 2,000 series). User decision (Stage D): query-time raw windows come from a local in-memory ring buffer, fed and sharded like ingestion and sized by the plan. Anything older comes from Prometheus remote read (protobuf), not the JSON API.
What
end_offset + widthof itsquery_time_raw_windows, and of its panes built at query time and kept (their retention, which covers the declared range evaluations). Metrics no raw window reads are not buffered.--raw-buffer-msoverrides the horizon; 0 keeps none. (A first commit used one horizon, the plan's longest read, for every metric. The second sizes it per metric, which is the plan's longest read applied to each metric; the table shows both.)/api/v1/read, snappy protobuf, sampled response). The JSON raw fetch is removed. Forwarding of unplanned queries is unchanged.How
RawBufferholds, per shard, the series' samples by metric and label set in time order, under a read-write lock, and trims each metric to its horizon. A read takes each shard's read lock, collects the samples in(start, end]and takes the latest trim point over the shards as the cut. Samples at or before the cut come from remote read, so a shard trimmed during the read cannot leave a gap.Before this PR
Raw windows: one JSON instant query per window or missing kept pane, decoded through
serde_json::Value.After this PR
A data plane that has received the samples since it started answers raw windows from memory, with no Prometheus request. Remote read covers only what predates its first sample or its horizon.
Verification
cargo test --workspacepasses, includinge2e_examples(10),answers,range,kept.e2e_examples::example1_raw_windows_from_the_buffer_and_remote_read, failing on feat(precompute): per-sample columnar ingestion and columnar raw-sample windows #830. Prometheus holds samples from 0 and the data plane receives them from 4:45. At 5 min the 1-minute window takes 4:15–4:45 from remote read and 5:00 from the buffer. At 10 min the window is the buffer's alone, with 0 remote reads. Every answer equals the Planner executor's over all samples, and the JSON API is never used for raw samples.raw_buffertests: reads equal the samples of the metric in(start, end]; a read past the horizon splits at the cut; samples before startup come from Prometheus.--raw-buffer-ms 0, so they count remote reads; the counts are unchanged (5, 1, 1 for B3; 1 per range query).Benefit harness
Same setup as #828. Each run writes every sample to the data plane from the start, so the buffer covers every raw window and no remote read happens.
s10
s10-noraw
s50
s50-noraw
"pr4-single-horizon" is this PR's first commit (one horizon for every metric); "pr4" is its head (per-metric horizons). "Prometheus alone" is the baseline of the pr3 run.
CPU saved against the JSON raw fetch (raw data retained; head against pr3):
Raw buffer size, at its largest, from the data plane's log. The plan's horizons are
http_requests_total1 min andlatency_ms60 min.The bytes are the samples' 16 bytes each; per-series map entries come on top. Peak RSS does not change measurably between pr3 and pr4.