refactor(emit): align typed L5 streaming-config JSON with backend parser - #244
Merged
Merged
Conversation
The typed L5 path's `emit_backend_streaming_config_json` was emitting
a stripped-down JSON shape that did not parse on the backend — every
`handle_plan` POST to `/api/v1/streaming-config` was failing with
HTTP 400 (`Missing grouping labels` / `Missing windowSize` / …). The
runtime tolerated this because the Replanner's separate legacy
emitter (`generate_streaming_config_yaml` in `emit/asapquery_backend.rs`)
posts valid YAML on each replan; but on a fresh deploy the backend had
no aggregations registered from the controller until the first replan.
This PR aligns the typed L5 wire shape with what
`asap_types::AggregationConfig::from_yaml_data` requires:
* Adds `metric_name`, `window_secs`, `spatial_filter` fields to
`BackendAggregation` (populated from the surrounding
`EdgeStageConfig` at the two construction sites — for
`SketchAgg @ Edge` and `RawAtEdgeSketchAtBackend @ Edge`).
* Rewrites `build_backend_aggregation_json` to emit the full
backend-expected shape: `aggregationType`, `aggregationSubType`,
`metric`, `labels.{grouping,rollup,aggregated}`, `parameters`,
`windowSize`, `windowType`, `spatialFilter`, `aggregationInput`.
PR 5 alignment: drops `aggregationId` from the wire entirely (it's
silently dropped by the backend parser anyway — identity is
content-addressed via `PolicyFingerprint(u64)` derived from the fields
above). The `aggregation_id` field remains on the struct as internal
emitter plumbing (`SketchAgg → BackendAggregation → BackendReadout`
cross-reference during the DAG walk), but never reaches the wire.
`labels.grouping` is emitted empty in this PR — the typed L5 doesn't
thread `QueryExpr::Aggregate.by` through `BackendAggregation` yet.
That's a follow-up; the parser accepts the empty list. `spatial_filter`
is populated from `edge.label_filters` (the `Scan` label filters
extracted by `extract_edge_facts`). `windowSize` comes from
`edge.window_secs` (the `Window` op extracted by the same).
Test updates:
- 9 fixtures updated to construct `BackendAggregation` with the new
fields.
- 2 assertions flipped from "aggregationId is present" to
"aggregationId is absent + metric/windowSize present".
- `backend_client.rs` mock-POST bodies updated to current shape.
- 1 helper added: `spatial_filter_from_label_filters`.
Build: clean. 686 lib + 27 binary tests pass.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
3 tasks
zzylol
added a commit
that referenced
this pull request
May 15, 2026
…els.grouping (#245) Follow-up to #244. The typed L5's `BackendAggregation` now carries a `grouping: Vec<String>` field, surfaced under `labels.grouping` in the streaming-config JSON the controller posts to asapquery-backend. The backend's precompute-engine accumulator pipeline keys its per-aggregation state by the projected attribute set, so this is what makes cross-host fan-in merges (sum by zone / etc.) actually behave correctly on the backend side. ## Plumbing rationale The L3 `QueryExpr::Aggregate.by` is `Vec<ColumnId>` — positional indexes into a synthesized `Schema` that intentionally does not track open-set labels (see the module doc on `intent_algebra::column_resolution` — label-set resolution is a Step γ TODO). So reverse-resolving ColumnId → label name at L5 emit time isn't tractable. `QueryWorkload.group_by_labels: Vec<String>` carries the names unambiguously, and `handle_plan` already has the workload in scope at the call site for both binder paths (query_string → `bind_query_expr` and explicit-field → `bind_workload_typed`). The pragmatic plumb: 1. Add `grouping: Vec<String>` to `BackendAggregation` (default empty; `#[serde(default)]` keeps existing serialised fixtures parsing). 2. Have the L5 emitter populate `grouping: vec![]` at both construction sites (no behaviour change at emit time). 3. In `handle_plan`, after `split_typed_three_stage` returns the per-stage map, patch every `BackendAggregation.grouping` with `workload.group_by_labels.clone()` before `emit_backend_streaming_config_json`. Every aggregation under a single workload shares the same grouping today, so the patch is uniform. 4. `build_backend_aggregation_json` reads `agg.grouping` into `labels.grouping` (was hard-coded empty before this PR). `labels.rollup` and `labels.aggregated` stay empty — the controller doesn't surface either today. ## Test plan - New: `backend_json_emits_grouping_under_labels` exercises the JSON round-trip of the new field directly. 687 lib tests (+1) + 27 binary tests pass. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
3 tasks
zzylol
added a commit
that referenced
this pull request
May 15, 2026
#246) Mirror of #244's aggregation-side cleanup, now for readouts. The backend's `StreamingConfig::from_yaml_data` doesn't consume the `readouts` list at all today, so this is wire-shape hygiene rather than a fix — but it keeps the controller's emit consistent with the content-addressing convention PR 5 established: no controller-allocated string IDs on the wire. When the backend eventually starts consuming readouts, the cross- reference from a readout to its source aggregation will be content- shaped (metric / sketch_kind / params) — derivable from the `aggregations` list by the same `PolicyFingerprint` recipe. The `BackendReadout.aggregation_id` struct field stays in source as emit-time bookkeeping (populated by `resolve_descendant_agg_id_via_edges` during the DAG walk; not consumed by anything downstream today). Touches: - `build_backend_readout_json`: 4 match arms each drop their `aggregationId` key. Doc-comment expanded to explain the content-addressing convention. - `phase_b_backend_json_aggregation_readout_alias_snapshot`: flips the readouts assertion from "aggregationId present" → "absent". Build clean. 687 lib + 27 binary tests pass. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
3 tasks
zzylol
added a commit
that referenced
this pull request
May 15, 2026
…g plumb (#247) First end-to-end test of the gateway-less control-plane ↔ data-plane contract. Lives in `data_plane/tests/` since data_plane already owns the OTLP/precompute/HTTP wiring; control_plane is already a workspace dep so no Cargo.toml changes were needed. ## What it tests Two `#[tokio::test]`s: 1. `controller_plans_ddsketch_quantile_and_backend_parses_streaming_config` — workload `{ metric: http_latency_ms, agg: Quantile, sla: 0.01, window: 60s }` flows through `bind_workload_typed` → `split_typed_three_stage` → `emit_backend_streaming_config_json`, the JSON is POSTed to the in-process backend's `/api/v1/streaming-config`, and the GET endpoint reflects the parsed aggregation. Asserts the emitted JSON's shape (#244 content fields present, #244/#246 `aggregationId` absent). 2. `controller_plans_with_grouping_and_backend_parses_grouping_labels` — same shape but with `group_by_labels: ["zone"]`. Asserts the emitter surfaces `labels.grouping: ["zone"]` (#245), the POST succeeds, and the active-config snapshot contains `"zone"` in the parsed `AggregationConfig.grouping_labels`. ## Findings the test caught `bind_workload_typed`-produced PhysicalExpr doesn't surface `metric_name` or `window_secs` to `BackendAggregation` via `extract_edge_facts`. The underlying cause is likely that the allocator paints the `Logical(Scan{...})` chain at a stage other than Edge in some binder outputs, so the walk's `(Logical, Edge)` match arm doesn't fire — Step γ open-set label work (also blocking grouping fidelity) probably needs a coordinated fix. Pragmatic workaround mirroring the #245 grouping patch: control_plane/src/main.rs (handle_plan, Backend stage match arm) for agg in &mut be.aggregations { if agg.metric_name.is_empty() { agg.metric_name = workload.metric_name.clone(); } if agg.window_secs == 0 { agg.window_secs = workload.time_window.as_secs(); } agg.grouping = workload.group_by_labels.clone(); } This is belt-and-braces — if the L5 walk DID surface the field, the conditional `if … is_empty()` / `== 0` checks preserve it; otherwise the workload spec wins. Closes a real bug where fresh-deploy `handle_plan` POSTs would have failed the backend parser on `Missing metric` / `Missing windowSize` (PR #244 caught the JSON shape gap; this test caught the value-population gap). ## Other change `HttpServer::start_test_server` had `#[cfg(test)]` gating it to the lib's own unit tests — integration tests under `data_plane/tests/` are compiled separately and couldn't see it. Dropped the gate; the method's name + doc-comment make the test-only intent explicit, and production code uses `start()` regardless. ## Test plan - [x] `cargo test --test e2e_controller_plans_and_backend_serves`: **2 passed**. - [x] `cargo test --lib -p control_plane`: **687 passed**. - [x] `cargo test --tests --bins -p control_plane`: **27 passed**. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2 tasks
zzylol
added a commit
that referenced
this pull request
May 15, 2026
…re (#251) Mirrors #244 / #246 / #250's PR-5 wire-cleanup pattern, now for the edge-side runtime emitters. The patched asap-otel processors (any runtime — OTel collector / OTAP / Telegraf) don't consume the controller-allocated `aggregation_id` string: sid identity is content-addressed at the backend via `(metric_name, attrs_fingerprint, agg_kind_canonical)`, and `policy_fp` content- matches via `(metric, sketch_kind, config, group_by_keys)`. The controller-allocated string is dead weight on the wire. The OTel-collector YAML emitter (`emit_edge_yaml`) was already clean — there's an existing test at stage_config.rs asserting the agent YAML does NOT contain `aggregation_id:`. This PR catches the two remaining edge-runtime emitters that still spelled it out: - `emit/otap.rs::build_asap_sketches_config` (line ~287): drop the `aggregation_id` map entry; keep `sketch_kind` etc. - `emit/telegraf.rs::emit_processors_allsketches` (line ~158): drop the `aggregation_id = "…"` TOML line; keep `sketch_kind`. `EdgeSketchProcessor.aggregation_id` stays on the struct as internal emitter plumbing for cross-stage references during the DAG walk (`SketchAgg → BackendAggregation → BackendReadout`); it just doesn't reach the wire from any runtime emitter anymore. Tests: 690 lib + 27 binary tests pass. (No test asserted the field was present in OTAP / Telegraf output, so nothing to flip.) Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
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.
Summary
The typed L5 path's
emit_backend_streaming_config_jsonwas emitting a stripped-down JSON shape that didn't parse on the backend — everyhandle_planPOST to/api/v1/streaming-configwas failing HTTP 400 (Missing grouping labels/Missing windowSize/Missing metric/ …). The deploy ran ”OK” only because the Replanner's separate legacy emitter (generate_streaming_config_yaml) posts valid YAML on each replan — but on a fresh deploy, the backend had no aggregations registered from the controller until the first replan fired.This PR closes the gap:
BackendAggregation—metric_name,window_secs,spatial_filter. Populated from the surroundingEdgeStageConfigat the two construction sites (SketchAgg @ Edge,RawAtEdgeSketchAtBackend @ Edge).build_backend_aggregation_jsonto emit the full backend-expected shape (aggregationType,aggregationSubType,metric,labels.{grouping,rollup,aggregated},parameters,windowSize,windowType,spatialFilter,aggregationInput).aggregationIdfrom the wire entirely (PR 5 alignment — the backend's parser silently dropped it anyway; identity is content-addressed viaPolicyFingerprint(u64)).aggregation_idremains on the struct as internal emitter plumbing for theSketchAgg → BackendAggregation → BackendReadoutcross-reference during DAG walk.ID design philosophy this PR commits to
Per the design discussion: the control plane emits content, not derived identifiers. The data_plane derives the
PolicyFingerprint(u64)from the same content; sids (per-series) are minted bySeriesIdResolverat ingest. The controller never participates in either u64 namespace — same content → same fingerprint, by construction.Out of scope (follow-ups)
labels.groupingis emitted empty in this PR. ThreadingQueryExpr::Aggregate.bythroughBackendAggregationrequires column-id-to-name resolution via the schema — separate concern. The backend's parser accepts the empty list, so this is correct-but-incomplete content fidelity.aggregationIdstays for now. The backend'sStreamingConfig::from_yaml_dataignores the entirereadoutslist today, so the field is informational. Cleanup belongs in whatever PR wires the readouts up.deploy/mvp-singlenode/configs/backend-streaming-{kll,cs,hll}.yaml) still carryaggregationId: Nlines — tiny separate PR.Test plan
cargo checkclean (only pre-existing warnings)cargo test --lib: 686 passed; 0 failedcargo test --tests --bins: 27 passed; 0 failed🤖 Generated with Claude Code