From 0e8c45b8af45f1faf3440695af71a57ed7bb21c3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 15 May 2026 13:37:45 -0600 Subject: [PATCH] refactor(emit): align typed L5 streaming-config JSON with backend parser MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- control_plane/src/backend_client.rs | 4 +- control_plane/src/emit/stage_config.rs | 95 ++++++++++++++++--- control_plane/src/emit/trait_def.rs | 3 + .../src/physical/colored_dag/emitter.rs | 35 ++++++- 4 files changed, 123 insertions(+), 14 deletions(-) diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 63f9bb3b..a71bb530 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -256,7 +256,7 @@ mod tests { let url = start_mock_backend(sink.clone(), axum::http::StatusCode::OK).await; let client = BackendClient::new(url); - let yaml = "aggregations:\n - aggregationId: 42\n metric: cpu\n".to_string(); + let yaml = "aggregations:\n - metric: cpu\n".to_string(); client .push_streaming_config(yaml.clone()) .await @@ -297,7 +297,7 @@ mod tests { let url = start_mock_backend(sink.clone(), axum::http::StatusCode::OK).await; let client = BackendClient::new(url); - let json = r#"{"aggregations":[{"aggregationId":7,"metric":"latency"}]}"#.to_string(); + let json = r#"{"aggregations":[{"metric":"latency"}]}"#.to_string(); client .post_streaming_config_json(json.clone()) .await diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 816253d3..48df7743 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -499,7 +499,10 @@ pub fn emit_gateway_yaml(cfg: &GatewayStageConfig, opamp_endpoint: &str) -> Resu /// Output shape mirrors the YAML shape produced by /// [`crate::config::asapquery_backend::generate_streaming_config_yaml`]: /// a top-level `aggregations` array of -/// `{ aggregationId, aggregationType, metric, parameters, ... }` rows. +/// `{ aggregationType, aggregationSubType, metric, labels, parameters, +/// windowSize, windowType, spatialFilter, aggregationInput }` rows. +/// `aggregationId` is **not** emitted — identity is content-addressed in +/// the backend via `PolicyFingerprint(u64)`. /// We additionally surface a parallel `readouts` array so the backend's /// query engine can prepare per-readout dispatch entries up-front (the /// existing YAML form has no readouts list because the legacy planner @@ -1357,21 +1360,49 @@ fn build_gateway_merge_block(mp: &GatewayMergeProcessor) -> Value { } /// Build one aggregation row in the backend streaming-config JSON. +/// +/// Wire shape is aligned to what `asap_types::AggregationConfig::from_yaml_data` +/// requires: +/// +/// * `aggregationType` — sketch family. +/// * `aggregationSubType` — always empty; reserved for future +/// sub-family distinctions. +/// * `metric` — source metric the aggregation runs over. +/// * `labels.{grouping,rollup,aggregated}` — three label lists the +/// backend's `KeyByLabelNames` parser keys on. Today the typed L5 +/// only surfaces an empty grouping; future work threads +/// `QueryExpr::Aggregate.by` through `BackendAggregation` so grouping +/// propagates faithfully. +/// * `parameters` — sketch-family-specific params (alpha, K, precision…). +/// * `windowSize` / `windowType` — tumbling window in seconds. +/// * `spatialFilter` — comma-joined `k=v` pairs from the edge's label +/// filters. +/// * `aggregationInput` — Phase ε.1 Mode 1/2 marker (sketch_envelope vs +/// raw); preserved so Phase ε.2's raw-input ingest path stays plumbed. +/// +/// `aggregation_id` is **intentionally omitted** from the wire — PR 5 +/// retired the controller-allocated id; identity is content-addressed +/// in the backend via `PolicyFingerprint(u64)` derived from the fields +/// above. fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue { let parameters = sketch_params_to_json(&agg.sketch_params); - // Phase ε.1 — surface `aggregation_input` so the backend's - // `StreamingConfig` consumer knows whether the wire payload is a - // pre-built sketch envelope (Mode 1) or raw OTLP samples the backend - // builds the sketch from at ingest (Mode 2). Phase ε.2 adds the - // raw-input ingest path; Phase ε.1 only commits the wire shape. let aggregation_input = match agg.aggregation_input { AggregationInput::SketchEnvelope => "sketch_envelope", AggregationInput::Raw => "raw", }; json!({ - "aggregationId": agg.aggregation_id, "aggregationType": sketch_kind_to_backend_type(&agg.sketch_kind), + "aggregationSubType": "", + "metric": agg.metric_name, + "labels": { + "grouping": Vec::::new(), + "rollup": Vec::::new(), + "aggregated": Vec::::new(), + }, "parameters": parameters, + "windowSize": agg.window_secs, + "windowType": "tumbling", + "spatialFilter": agg.spatial_filter, "aggregationInput": aggregation_input, }) } @@ -1708,14 +1739,20 @@ mod tests { aggregations: vec![ BackendAggregation { aggregation_id: "agg0".into(), + metric_name: "http_latency_ms".into(), sketch_kind: SketchKind::DDSketch, sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::SketchEnvelope, }, BackendAggregation { aggregation_id: "agg1".into(), + metric_name: "http_requests_total".into(), sketch_kind: SketchKind::Hll, sketch_params: SketchParams::Hll(HllParams { precision: 14 }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::SketchEnvelope, }, ], @@ -1734,10 +1771,19 @@ mod tests { let aggs = v["aggregations"].as_array().expect("aggregations array"); assert_eq!(aggs.len(), 2, "{v}"); - assert_eq!(aggs[0]["aggregationId"], "agg0"); + // PR 5: `aggregationId` is no longer on the wire — identity is + // content-addressed in the backend via `PolicyFingerprint(u64)`. + assert!( + aggs[0].get("aggregationId").is_none(), + "controller must not emit aggregationId\n{v}" + ); assert_eq!(aggs[0]["aggregationType"], "DDSketch"); + assert_eq!(aggs[0]["metric"], "http_latency_ms"); + assert_eq!(aggs[0]["windowSize"], 60); + assert_eq!(aggs[0]["windowType"], "tumbling"); assert_eq!(aggs[0]["parameters"]["alpha"], 0.01); assert_eq!(aggs[1]["aggregationType"], "HLL"); + assert_eq!(aggs[1]["metric"], "http_requests_total"); assert_eq!(aggs[1]["parameters"]["precision"], 14); let reads = v["readouts"].as_array().expect("readouts array"); @@ -1753,18 +1799,24 @@ mod tests { aggregations: vec![ BackendAggregation { aggregation_id: "agg0".into(), + metric_name: "endpoint_count".into(), sketch_kind: SketchKind::CountSketch, sketch_params: SketchParams::CountSketch(CountSketchParams { w: 2048, d: 5, with_heap: true, }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::SketchEnvelope, }, BackendAggregation { aggregation_id: "agg1".into(), + metric_name: "endpoint_hits".into(), sketch_kind: SketchKind::Cms, sketch_params: SketchParams::Cms(CmsParams { w: 4096, d: 4 }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::SketchEnvelope, }, ], @@ -1823,8 +1875,11 @@ mod tests { BackendStageConfig { aggregations: vec![BackendAggregation { aggregation_id: "agg0".into(), + metric_name: "test_metric".into(), sketch_kind: kind.clone(), sketch_params: params, + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::SketchEnvelope, }], readouts: vec![BackendReadout { @@ -2182,8 +2237,11 @@ mod tests { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { aggregation_id: "phase_b_agg0".into(), + metric_name: "phase_b_metric".into(), sketch_kind: SketchKind::Kll, sketch_params: SketchParams::Kll(KllParams { k: 200 }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::SketchEnvelope, }], readouts: vec![BackendReadout { @@ -2192,9 +2250,18 @@ mod tests { }], }; let v = emit_backend_streaming_config_json(&cfg).expect("emit ok"); - // The id surfaces on both the agg and the readout, with the same - // key name — the backend looks the readout up by `aggregationId`. - assert_eq!(v["aggregations"][0]["aggregationId"], "phase_b_agg0"); + // PR 5: `aggregationId` is no longer on the aggregation side — the + // backend derives identity from content (`PolicyFingerprint(u64)` + // over metric, sketch_kind, params, grouping, spatial_filter). + // Readouts still surface `aggregationId` because the backend's + // readout consumption path is unchanged (cleanup deferred — the + // current backend's `StreamingConfig::from_yaml_data` ignores the + // readouts list entirely, so this string is informational only). + assert!( + v["aggregations"][0].get("aggregationId").is_none(), + "controller must not emit aggregationId on aggregations\n{v}" + ); + assert_eq!(v["aggregations"][0]["metric"], "phase_b_metric"); assert_eq!(v["readouts"][0]["aggregationId"], "phase_b_agg0"); assert_eq!(v["aggregations"][0]["aggregationType"], "DatasketchesKLL"); assert_eq!(v["aggregations"][0]["parameters"]["k"], 200); @@ -2212,8 +2279,11 @@ mod tests { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { aggregation_id: "agg0".into(), + metric_name: "test_metric".into(), sketch_kind: SketchKind::DDSketch, sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::SketchEnvelope, }], readouts: vec![], @@ -2231,8 +2301,11 @@ mod tests { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { aggregation_id: "agg0".into(), + metric_name: "test_metric".into(), sketch_kind: SketchKind::DDSketch, sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: AggregationInput::Raw, }], readouts: vec![], diff --git a/control_plane/src/emit/trait_def.rs b/control_plane/src/emit/trait_def.rs index 267e65af..9cd29a91 100644 --- a/control_plane/src/emit/trait_def.rs +++ b/control_plane/src/emit/trait_def.rs @@ -218,8 +218,11 @@ mod tests { BackendStageConfig { aggregations: vec![BackendAggregation { aggregation_id: "agg0".to_string(), + metric_name: "test_metric".to_string(), sketch_kind: SketchKind::DDSketch, sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + window_secs: 60, + spatial_filter: String::new(), aggregation_input: crate::physical::colored_dag::emitter::AggregationInput::SketchEnvelope, }], diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index a2a314c9..e08c2dc9 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -283,16 +283,34 @@ pub struct BackendStageConfig { } /// One sketch source the backend must accept. +/// +/// `aggregation_id` is **internal plumbing only** — used by the emitter +/// to thread `SketchAgg` → `BackendAggregation` → `BackendReadout` +/// during the DAG walk. It is **not** emitted on the wire (PR 5 retired +/// the controller-allocated id; the backend content-addresses identity +/// via `PolicyFingerprint(u64)` derived from `metric_name`, +/// `sketch_kind`, `sketch_params`, grouping labels, and `spatial_filter`). #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct BackendAggregation { - /// Stable id matching the upstream gateway's `aggregation_id`. + /// Internal-only id (see struct doc). Not on the wire. pub aggregation_id: String, + /// Source metric the aggregation runs over (e.g. + /// `http_requests_total_latency_ms`). Required by the backend's + /// `AggregationConfig` parser. + pub metric_name: String, /// Sketch family. pub sketch_kind: SketchKind, /// Sketch parameters — the backend uses these to build its /// per-aggregation `Sketch` instance (KLL with the right `k`, /// DDSketch with the right `alpha`, etc.). pub sketch_params: SketchParams, + /// Tumbling window size in seconds. Required by the backend; the + /// parser rejects zero-window aggregations. + pub window_secs: u64, + /// Spatial filter (comma-joined `k=v` pairs from the edge's + /// `label_filters`). Empty string when no filter applies. + #[serde(default)] + pub spatial_filter: String, /// Phase ε.1 — what shape the backend ingests for this /// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the /// default (the wire payload is a sketch state already). Mode 2 @@ -444,8 +462,11 @@ impl Emitter for ThreeStageEmitter { }); backend_aggregations.push(BackendAggregation { aggregation_id, + metric_name: edge.source_metric.clone().unwrap_or_default(), sketch_kind: sketch_type.clone(), sketch_params: params.clone(), + window_secs: edge.window_secs.unwrap_or(0), + spatial_filter: spatial_filter_from_label_filters(&edge.label_filters), // Mode 1 — sketch built at edge, ships envelope. aggregation_input: AggregationInput::SketchEnvelope, }); @@ -518,8 +539,11 @@ impl Emitter for ThreeStageEmitter { next_agg_index += 1; backend_aggregations.push(BackendAggregation { aggregation_id: aid, + metric_name: edge.source_metric.clone().unwrap_or_default(), sketch_kind: family.clone(), sketch_params: params.clone(), + window_secs: edge.window_secs.unwrap_or(0), + spatial_filter: spatial_filter_from_label_filters(&edge.label_filters), // Mode 2 — backend builds sketch from raw OTLP. aggregation_input: AggregationInput::Raw, }); @@ -637,6 +661,15 @@ fn extract_edge_facts(qe: &crate::intent_algebra::QueryExpr, edge: &mut EdgeStag } } +/// Join `label_filters` into the comma-separated `k=v` form the backend's +/// `AggregationConfig` spatial-filter parser accepts. Empty list → empty +/// string (the backend reads that as "no spatial filter"). +pub(crate) fn spatial_filter_from_label_filters(filters: &[(String, String)]) -> String { + let mut parts: Vec = filters.iter().map(|(k, v)| format!("{k}={v}")).collect(); + parts.sort(); + parts.join(",") +} + /// Children of `parent` per the DAG's edges table. The colouring walker /// emits parent → child edges in visit order so this iterator is /// deterministic.