diff --git a/control_plane/src/emit/asapquery_backend.rs b/control_plane/src/emit/asapquery_backend.rs deleted file mode 100644 index f072a5ad..00000000 --- a/control_plane/src/emit/asapquery_backend.rs +++ /dev/null @@ -1,287 +0,0 @@ -//! Convert a [`CollectionPlan`] into the YAML shape ASAPQuery-backend's -//! `POST /api/v1/streaming-config` endpoint accepts (the same format its -//! `StreamingConfig::from_yaml_data` parser consumes at startup). -//! -//! This is **separate** from `config::backend` (which produces OTel YAML -//! for a backend OTel collector running merge processors). The two -//! consumers are different: -//! -//! * `config::backend` — OTel collector, expects -//! `processors: { ddsketch_merge: {...} }` + `service.pipelines`. -//! * `config::asapquery_backend` (this module) — ASAPQuery-backend -//! query engine, expects -//! `aggregations: [{ aggregationType, metric, labels, parameters, -//! windowSize, windowType, spatialFilter }]`. -//! -//! Both are generated from the same `CollectionPlan` fields but target -//! different services. The replanner pushes the OTel YAML via OpAMP to -//! backend-role collectors and pushes this one via HTTP to the -//! ASAPQuery-backend's `/api/v1/streaming-config` endpoint. -//! -//! Phase 5 M2.2: this emitter no longer writes `aggregationId`. The -//! backend's `AggregationConfig::from_yaml_data` derives one -//! deterministically via `compute_agg_config_id` from the same set of -//! fields we emit, so the explicit field is redundant. - -use std::time::Duration; - -use anyhow::{Context, Result}; - -use crate::types::{AgentCollectorConfig, CollectionPlan, SketchType}; - -/// Generate the `StreamingConfig` YAML for the ASAPQuery-backend from a -/// single-metric `CollectionPlan`. Produces a one-element `aggregations` -/// list — the backend's endpoint will merge this into its active config -/// (add on conflict, replace on same id). -/// -/// # Errors -/// -/// Returns an error if the plan is missing a window (the backend's -/// config parser rejects zero-window aggregations) or if YAML -/// serialization fails. -pub fn generate_streaming_config_yaml(metric: &str, plan: &CollectionPlan) -> Result { - let agg = &plan.agent_config; - // MVP blocker B4: clamp the workload's `window_duration` to - // `[MIN_WINDOW_SECS, MAX_WINDOW_SECS]` so the legacy YAML emit - // matches the typed L5 JSON emit's `windowSize` clamp. Without - // this, the legacy and typed paths can disagree (e.g. typed - // clamps `[5m]` → 60, legacy passes 300 → backend reducer keys - // a 300s window the agent never closes). - let window_secs = agg - .window_duration - .map(|d: Duration| { - super::stage_config::clamp_window_secs(Some(d.as_secs())) - .expect("clamp preserves Some") - }) - .unwrap_or(0); - if window_secs == 0 { - anyhow::bail!( - "generate_streaming_config_yaml: plan for metric {metric} has \ - no window_duration; ASAPQuery-backend rejects zero-window aggregations" - ); - } - - // Parameters map: copy sketch-type-specific params (K for KLL, - // epsilon/delta for CountMin, etc.) into the string-keyed YAML map - // the backend expects. We serialize via serde_yaml to pick up - // SketchParams' own Serialize impl and then re-parse into a - // generic Mapping so we can embed it. - let params_yaml = serde_yaml::to_value(&agg.sketch_params) - .context("serialize SketchParams for ASAPQuery streaming config")?; - - let agg_type_str = map_sketch_type_to_agg_type(&agg.sketch_type); - - let aggregation = serde_yaml::Mapping::from_iter([ - ( - serde_yaml::Value::from("aggregationType"), - serde_yaml::Value::from(agg_type_str), - ), - ( - serde_yaml::Value::from("aggregationSubType"), - serde_yaml::Value::from(""), - ), - ( - serde_yaml::Value::from("metric"), - serde_yaml::Value::from(metric), - ), - ( - serde_yaml::Value::from("labels"), - labels_mapping(&agg.aggregate_by), - ), - (serde_yaml::Value::from("parameters"), params_yaml), - ( - serde_yaml::Value::from("windowSize"), - serde_yaml::Value::from(window_secs), - ), - ( - serde_yaml::Value::from("windowType"), - serde_yaml::Value::from("tumbling"), - ), - ( - serde_yaml::Value::from("spatialFilter"), - serde_yaml::Value::from(normalize_spatial_filter(&agg.label_matchers)), - ), - ]); - - let top = serde_yaml::Mapping::from_iter([( - serde_yaml::Value::from("aggregations"), - serde_yaml::Value::Sequence(vec![serde_yaml::Value::Mapping(aggregation)]), - )]); - - serde_yaml::to_string(&serde_yaml::Value::Mapping(top)) - .context("serialize ASAPQuery streaming-config YAML") -} - -/// Map the control plane's `SketchType` to the backend's -/// `AggregationType::Display` string. These strings must match what the -/// backend's `FromStr for AggregationType` in -/// `promql_utilities::query_logics::enums` accepts — hence the variant -/// names rather than the collector factory names (e.g. `"DatasketchesKLL"` -/// not `"KLL"`). -fn map_sketch_type_to_agg_type(t: &SketchType) -> &'static str { - match t { - SketchType::DDSketch => "DDSketch", - SketchType::KLL => "DatasketchesKLL", - SketchType::HLL => "HLL", - SketchType::CountSketch => "CountSketch", - SketchType::CountMinSketch => "CountMinSketch", - } -} - -/// Build the `labels` sub-mapping the backend expects. All three lists -/// exist because the backend's parser reads them separately for -/// key-value / spatial-rollup distinction; today the control plane only -/// tracks `aggregate_by` (grouping), so rollup and aggregated stay -/// empty and are populated in a follow-up when the cost model starts -/// producing richer label metadata. -fn labels_mapping(aggregate_by: &[String]) -> serde_yaml::Value { - serde_yaml::Value::Mapping(serde_yaml::Mapping::from_iter([ - ( - serde_yaml::Value::from("grouping"), - serde_yaml::Value::Sequence( - aggregate_by - .iter() - .cloned() - .map(serde_yaml::Value::from) - .collect(), - ), - ), - ( - serde_yaml::Value::from("rollup"), - serde_yaml::Value::Sequence(vec![]), - ), - ( - serde_yaml::Value::from("aggregated"), - serde_yaml::Value::Sequence(vec![]), - ), - ])) -} - -/// Join the controller's `label_matchers` list (each shaped like -/// `"key=value"`) into a single comma-separated string the backend's -/// spatial-filter parser accepts. When the list is empty, returns an -/// empty string (the backend treats that as "no spatial filter"). -fn normalize_spatial_filter(label_matchers: &[String]) -> String { - label_matchers.join(",") -} - -// ─── Unused-warning suppression for types that are referenced only -// inside the unit tests below. This keeps the module self-contained -// even when the rest of the controller crate's cfg(test) surface grows. -#[allow(dead_code)] -fn _type_check(_: &AgentCollectorConfig) {} - -#[cfg(test)] -mod tests { - use super::*; - use crate::types::{ - AgentDataSink, CollectionPlan, DeltaDecision, GatewayCollectorConfig, OutputMode, - ProcessorMode, SketchParams, TransmissionCostSummary, - }; - use std::time::Duration; - - fn dummy_plan(sketch_type: SketchType) -> CollectionPlan { - CollectionPlan { - agent_config: AgentCollectorConfig { - output_mode: OutputMode::Sketch, - sketch_type: sketch_type.clone(), - sketch_params: SketchParams::default(), - aggregate_by: vec!["host".to_string(), "service".to_string()], - label_matchers: vec!["env=prod".to_string()], - window_duration: Some(Duration::from_secs(30)), - mode: ProcessorMode::Window, - enable_self_monitoring: false, - transmit_sketch: true, - drop_original: true, - enable_series_id: false, - series_id_ttl_secs: 0, - delta_transmission: false, - delta_threshold: 0.0, - data_sink: AgentDataSink::default(), - }, - gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], - valid_until: chrono::Utc::now() + chrono::Duration::seconds(300), - delta_decision: DeltaDecision::default(), - transmission_cost_summary: TransmissionCostSummary::default(), - } - } - - #[test] - fn emitted_yaml_omits_aggregation_id() { - let plan = dummy_plan(SketchType::DDSketch); - let yaml = generate_streaming_config_yaml("cpu_usage", &plan).expect("yaml ok"); - let parsed: serde_yaml::Value = serde_yaml::from_str(&yaml).expect("re-parse ok"); - let a = &parsed["aggregations"][0]; - assert!( - a["aggregationId"].is_null(), - "controller must not emit aggregationId — backend derives it from content (M2.2)" - ); - } - - #[test] - fn yaml_round_trips_through_serde_yaml() { - let plan = dummy_plan(SketchType::DDSketch); - let yaml = generate_streaming_config_yaml("cpu_usage", &plan).expect("yaml ok"); - let parsed: serde_yaml::Value = serde_yaml::from_str(&yaml).expect("re-parse ok"); - - let aggs = parsed["aggregations"].as_sequence().expect("sequence"); - assert_eq!(aggs.len(), 1); - let a = &aggs[0]; - assert_eq!(a["aggregationType"], serde_yaml::Value::from("DDSketch")); - assert_eq!(a["metric"], serde_yaml::Value::from("cpu_usage")); - assert_eq!(a["windowSize"], serde_yaml::Value::from(30u64)); - assert_eq!(a["windowType"], serde_yaml::Value::from("tumbling")); - assert_eq!(a["spatialFilter"], serde_yaml::Value::from("env=prod")); - - let grouping = a["labels"]["grouping"].as_sequence().expect("grouping seq"); - let grouping: Vec<&str> = grouping.iter().filter_map(|v| v.as_str()).collect(); - assert_eq!(grouping, vec!["host", "service"]); - } - - #[test] - fn maps_all_sketch_types() { - assert_eq!( - map_sketch_type_to_agg_type(&SketchType::DDSketch), - "DDSketch" - ); - assert_eq!( - map_sketch_type_to_agg_type(&SketchType::KLL), - "DatasketchesKLL", - "KLL must map to the backend's enum variant name, not the factory name" - ); - assert_eq!(map_sketch_type_to_agg_type(&SketchType::HLL), "HLL"); - assert_eq!( - map_sketch_type_to_agg_type(&SketchType::CountSketch), - "CountSketch" - ); - assert_eq!( - map_sketch_type_to_agg_type(&SketchType::CountMinSketch), - "CountMinSketch" - ); - } - - #[test] - fn rejects_plan_without_window_duration() { - let mut plan = dummy_plan(SketchType::HLL); - plan.agent_config.window_duration = None; - let err = generate_streaming_config_yaml("m", &plan).expect_err("should error"); - assert!( - err.to_string().contains("window_duration"), - "error should mention window_duration: {err}" - ); - } - - #[test] - fn spatial_filter_joins_label_matchers() { - let mut plan = dummy_plan(SketchType::DDSketch); - plan.agent_config.label_matchers = - vec!["env=prod".to_string(), "region=us-east".to_string()]; - let yaml = generate_streaming_config_yaml("m", &plan).expect("ok"); - let parsed: serde_yaml::Value = serde_yaml::from_str(&yaml).unwrap(); - assert_eq!( - parsed["aggregations"][0]["spatialFilter"], - serde_yaml::Value::from("env=prod,region=us-east") - ); - } -} diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs new file mode 100644 index 00000000..e70387bf --- /dev/null +++ b/control_plane/src/emit/backend_push.rs @@ -0,0 +1,306 @@ +//! Typed cumulative push of `BackendStageConfig` to the ASAPQuery-backend. +//! +//! Single entrypoint — [`post_typed_backend_for_role`] — invoked from +//! every plan-emit cycle (HTTP `POST /api/v1/plan`, the replanner's +//! plan-expiry / SLA-violation triggers, startup pre-pop tick, OpAMP +//! on-connect tick). It: +//! +//! 1. Updates the per-`(metric, role)` cache with the new +//! `BackendStageConfig`. +//! 2. Builds a **cumulative** `BackendStageConfig` whose +//! `aggregations` + `readouts` concatenate every cache entry's, +//! ordered deterministically (`(metric, role.as_str())` ascending) +//! so the emitted JSON body is reproducible across runs and tests. +//! 3. POSTs the cumulative streaming-config JSON to +//! `/api/v1/streaming-config` — the data plane's atomic +//! `handle.swap(new_config)` then installs every role's +//! aggregations simultaneously. +//! 4. Groups cache entries by metric, merges each metric's +//! `BackendStageConfig`s, and POSTs the per-metric merged routing +//! table to `/api/v1/storage_routing`. +//! +//! **Why one helper, not two paths**: prior to Option B the control +//! plane had two emit paths into the backend: +//! +//! * the typed cumulative path from `handle_plan` (post PR #287) — +//! correct under the data plane's swap semantics; +//! * the legacy single-aggregation path from `Replanner` +//! (`generate_streaming_config_yaml`) — emits ONE aggregation per +//! POST. Under the swap, this WIPES the cumulative state on the +//! backend the moment plan-expiry or accuracy-violation fires it. +//! +//! Option B unifies both call sites through this helper so the swap +//! semantics are honoured at every emit cycle, and the legacy YAML +//! emitter is retired. +//! +//! Fire-and-forget contract: every error (emit failure, HTTP transport +//! error, non-2xx response) logs at WARN and returns — never panics, +//! never propagates. The next replan cycle retries with the latest +//! plan. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; + +use tokio::sync::Mutex; +use tracing::{info, warn}; + +use crate::backend_client::BackendClient; +use crate::emit::{emit_backend_storage_routing, emit_backend_streaming_config_json}; +use crate::physical::colored_dag::emitter::BackendStageConfig; +use crate::workload::AggRole; + +/// Per-`(metric, role)` `BackendStageConfig` cache type alias. The +/// cache is owned by the controller's `AppState` and shared with the +/// `Replanner` via `Arc>` so both call sites read/write the +/// same cumulative state. +pub type BackendRoutingCache = Mutex>; + +/// Update the cumulative cache with `be` for `(metric, role)` and +/// POST the cumulative streaming-config + storage-routing JSON +/// documents to the backend. +/// +/// `backend_client`: `None` is the explicit "no backend configured" +/// signal — the function still logs the would-have-emitted shape and +/// returns. This preserves the fire-and-forget contract from PR #287. +/// +/// Errors at any step are logged at WARN and returned — never +/// propagated. +pub async fn post_typed_backend_for_role( + backend_client: Option<&Arc>, + cache: &BackendRoutingCache, + metric: &str, + role: AggRole, + be: BackendStageConfig, +) { + // ── 1. Update cache and collect cumulative entries ─────────────────── + // + // Snapshot the cache under a single lock so concurrent calls don't + // interleave half-applied state. The clone is cheap (the per-(metric, + // role) BackendStageConfig payloads are O(aggregations + readouts) + // and one POST cycle). + let cumulative_entries: Vec<((String, AggRole), BackendStageConfig)> = { + let mut cache = cache.lock().await; + cache.insert((metric.to_string(), role), be); + let mut v: Vec<((String, AggRole), BackendStageConfig)> = cache + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + // Deterministic ordering so the emitted JSON body is + // reproducible across runs (HashMap iteration would otherwise + // make captured-body regression assertions flaky). + v.sort_by(|(a_k, _), (b_k, _)| { + a_k.0 + .cmp(&b_k.0) + .then_with(|| a_k.1.as_str().cmp(b_k.1.as_str())) + }); + v + }; + + // ── 2. Cumulative streaming-config ─────────────────────────────────── + // + // One `BackendStageConfig` whose `aggregations` + `readouts` are the + // concatenation of every cache entry's. The data plane's swap + // installs this single multi-aggregation config atomically, so ALL + // roles for ALL metrics survive. + let cumulative_be = BackendStageConfig { + aggregations: cumulative_entries + .iter() + .flat_map(|(_, c)| c.aggregations.iter().cloned()) + .collect(), + readouts: cumulative_entries + .iter() + .flat_map(|(_, c)| c.readouts.iter().cloned()) + .collect(), + }; + + match emit_backend_streaming_config_json(&cumulative_be) { + Ok(json_doc) => { + info!( + stage = "backend", + metric = %metric, + role = %role, + aggregations = cumulative_be.aggregations.len(), + readouts = cumulative_be.readouts.len(), + cumulative_pairs = cumulative_entries.len(), + "[USE_TYPED_STAGE_SPLIT] posting typed backend JSON" + ); + if let Some(client) = backend_client { + let body = json_doc.to_string(); + match client.post_streaming_config_json(body).await { + Ok(()) => info!( + stage = "backend", + endpoint = %client.endpoint(), + "[USE_TYPED_STAGE_SPLIT] typed backend JSON push succeeded" + ), + Err(e) => warn!( + stage = "backend", + endpoint = %client.endpoint(), + error = %e, + "[USE_TYPED_STAGE_SPLIT] typed backend JSON push failed; \ + next replan cycle will retry" + ), + } + } else { + info!( + stage = "backend", + "[USE_TYPED_STAGE_SPLIT] no backend client configured; \ + skipping JSON push (set CONTROLLER_BACKEND_ENDPOINT to enable)" + ); + } + } + Err(e) => warn!(error = %e, "emit_backend_streaming_config_json failed"), + } + + // ── 3. Cumulative storage-routing ──────────────────────────────────── + // + // The routing classifier (`build_routing_entry` in + // `emit/stage_config.rs`) reads `cfg.aggregations` to derive shape + // routing, so we MUST merge every role's aggregations for one + // metric into a single `BackendStageConfig` before passing it + // through — otherwise a metric with both DDSketch (Quantile) and + // ExactAgg (Sum) would emit only the last-cached role's shape + // classifications and route the siblings to archive. + // + // `emit_backend_storage_routing`'s signature is + // `&[(String, &BackendStageConfig)]` — per-metric, NOT per-(metric, + // role) — so the merge happens here. + let mut by_metric: BTreeMap = BTreeMap::new(); + for ((m, _r), cfg) in &cumulative_entries { + let entry = by_metric.entry(m.clone()).or_insert_with(|| { + BackendStageConfig { + aggregations: Vec::new(), + readouts: Vec::new(), + } + }); + entry.aggregations.extend(cfg.aggregations.iter().cloned()); + entry.readouts.extend(cfg.readouts.iter().cloned()); + } + let routing_owned: Vec<(String, BackendStageConfig)> = by_metric.into_iter().collect(); + let routing_input: Vec<(String, &BackendStageConfig)> = routing_owned + .iter() + .map(|(k, v)| (k.clone(), v)) + .collect(); + match emit_backend_storage_routing(&routing_input) { + Ok(routing_doc) => { + info!( + stage = "backend", + metric = %metric, + cumulative_metrics = routing_owned.len(), + cumulative_pairs = cumulative_entries.len(), + "[USE_TYPED_STAGE_SPLIT] posting cumulative storage-routing JSON" + ); + if let Some(client) = backend_client { + let body = routing_doc.to_string(); + match client.post_storage_routing_json(body).await { + Ok(()) => info!( + stage = "backend", + metric = %metric, + "[USE_TYPED_STAGE_SPLIT] storage-routing JSON push succeeded" + ), + Err(e) => warn!( + stage = "backend", + metric = %metric, + error = %e, + "[USE_TYPED_STAGE_SPLIT] storage-routing JSON push failed; \ + next replan cycle will retry" + ), + } + } else { + info!( + stage = "backend", + "[USE_TYPED_STAGE_SPLIT] no backend client configured; \ + skipping storage-routing JSON push" + ); + } + } + Err(e) => warn!(error = %e, "emit_backend_storage_routing failed"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_be(metric: &str, agg_id: &str) -> BackendStageConfig { + use crate::physical::colored_dag::emitter::{ + AggregationInput, BackendAggregation, BackendReadout, + }; + use crate::sketch_algebra::params::{DDSketchParams, SketchKind, SketchParams}; + use crate::sketch_algebra::physical_expr::EstimateOp; + BackendStageConfig { + aggregations: vec![BackendAggregation { + aggregation_id: agg_id.to_string(), + metric_name: metric.to_string(), + sketch_kind: SketchKind::DDSketch, + sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + grouping: vec![], + spatial_filter: String::new(), + window_secs: 60, + aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, + }], + readouts: vec![BackendReadout { + aggregation_id: agg_id.to_string(), + op: EstimateOp::Quantile { q: 0.99 }, + }], + } + } + + /// With no backend client, the helper logs but returns without + /// panic. The cache is still updated (verified by a second call + /// asserting cumulative behaviour). + #[tokio::test] + async fn no_client_still_updates_cache() { + let cache = Mutex::new(HashMap::new()); + let be = make_be("m", "agg0"); + post_typed_backend_for_role(None, &cache, "m", AggRole::Quantile, be).await; + let snap = cache.lock().await; + assert_eq!(snap.len(), 1); + assert!(snap.contains_key(&("m".to_string(), AggRole::Quantile))); + } + + /// Two distinct `(metric, role)` calls produce a cumulative cache + /// (size 2), not overwrite (size 1). This is the regression that + /// motivated PR #287. + #[tokio::test] + async fn distinct_roles_accumulate_not_overwrite() { + let cache = Mutex::new(HashMap::new()); + post_typed_backend_for_role( + None, + &cache, + "http_requests_total", + AggRole::Quantile, + make_be("http_requests_total", "q"), + ) + .await; + post_typed_backend_for_role( + None, + &cache, + "http_requests_total", + AggRole::Sum, + make_be("http_requests_total", "s"), + ) + .await; + let snap = cache.lock().await; + assert_eq!(snap.len(), 2, "both roles must persist"); + assert!(snap.contains_key(&("http_requests_total".to_string(), AggRole::Quantile))); + assert!(snap.contains_key(&("http_requests_total".to_string(), AggRole::Sum))); + } + + /// Re-posting the same `(metric, role)` is idempotent at the + /// cache level — the entry is replaced, not duplicated. This is + /// the contract OpAMP on-connect ticks rely on (multiple + /// reconnects must not bloat the cumulative POST). + #[tokio::test] + async fn same_pair_replaces_not_duplicates() { + let cache = Mutex::new(HashMap::new()); + post_typed_backend_for_role(None, &cache, "m", AggRole::Quantile, make_be("m", "v1")) + .await; + post_typed_backend_for_role(None, &cache, "m", AggRole::Quantile, make_be("m", "v2")) + .await; + let snap = cache.lock().await; + assert_eq!(snap.len(), 1); + let entry = snap.get(&("m".to_string(), AggRole::Quantile)).unwrap(); + assert_eq!(entry.aggregations[0].aggregation_id, "v2"); + } +} diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index b9d6230f..3946eca8 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -8,7 +8,8 @@ //! |---|---| //! | `config/agent.rs` | [`agent`] | //! | `config/backend.rs` | *retired — emitted YAML for a "backend-role" OTel merge collector tier that was never deployed; superseded by the typed L5's [`stage_config::emit_backend_streaming_config_json`] which posts to asapquery-backend's precompute engine over HTTP* | -//! | `config/asapquery_backend.rs` | [`asapquery_backend`] | +//! | `config/asapquery_backend.rs` | *retired — `generate_streaming_config_yaml` was the legacy single-aggregation `CollectionPlan`-shaped emitter for `POST /api/v1/streaming-config`; under the data plane's atomic `handle.swap(new_config)` it would WIPE sibling `(metric, role)` aggregations on every fire. Replaced by [`backend_push::post_typed_backend_for_role`], which posts a cumulative typed `BackendStageConfig` derived from the shared per-`(metric, role)` cache* | +//! | *(new)* | [`backend_push`] | //! | `config/precompute.rs` | [`precompute`] | //! | `config/stage_config.rs` | [`stage_config`] (TODO: split into `opamp` + `streaming_config` + `inference_config` per design.md §5; deferred from refactor 2026-05 because the 3,020-line monolith mixes OTel-collector YAML emit, ASAPQuery-backend JSON emit, and shared internals — clean split needs ownership reorganisation, not file renames) | //! | `config/stage_config_otap.rs` | [`otap`] | @@ -16,7 +17,7 @@ //! | `config/workloads.rs` | [`crate::workload`] (top-level — design.md §5 puts `workload` next to `emit`, not inside it) | pub mod agent; -pub mod asapquery_backend; +pub mod backend_push; pub mod otap; pub mod precompute; pub mod stage_config; @@ -24,7 +25,7 @@ pub mod telegraf; pub mod trait_def; pub use agent::generate_agent_collector_config; -pub use asapquery_backend::generate_streaming_config_yaml; +pub use backend_push::{post_typed_backend_for_role, BackendRoutingCache}; pub use otap::emit_otap_dag_yaml; pub use precompute::{build_precompute_engine_jobs, should_precompute, PrecomputeClient}; pub use stage_config::{ diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 866c3dde..2c816e16 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -8,10 +8,11 @@ //! aggregator (OTLP receiver → per-family `*merge` processor(s) → OTLP //! exporter to backend). //! - [`emit_backend_streaming_config_json`] → JSON document matching the -//! ASAPQuery-backend `POST /api/v1/streaming-config` API surface — same -//! shape that [`crate::config::asapquery_backend::generate_streaming_config_yaml`] -//! builds today, just from the typed [`BackendStageConfig`] instead of -//! a `CollectionPlan`. +//! ASAPQuery-backend `POST /api/v1/streaming-config` API surface, +//! sourced from the typed [`BackendStageConfig`]. The legacy +//! `generate_streaming_config_yaml` `CollectionPlan`-shaped emitter +//! was retired in the Option B unification (see +//! [`crate::emit::backend_push`]). //! - [`emit_backend_storage_routing`] → JSON document matching the //! ASAPQuery-backend `POST /api/v1/storage_routing` API surface — //! per-metric query-shape → engine routing table (Phase α). Sources @@ -584,11 +585,12 @@ pub fn emit_gateway_yaml( /// `POST /api/v1/streaming-config` endpoint accepts, sourced from the /// typed L5 [`BackendStageConfig`]. /// -/// Output shape mirrors the YAML shape produced by -/// [`crate::config::asapquery_backend::generate_streaming_config_yaml`]: -/// a top-level `aggregations` array of +/// Output shape: a top-level `aggregations` array of /// `{ aggregationType, aggregationSubType, metric, labels, parameters, /// windowSize, windowType, spatialFilter, aggregationInput }` rows. +/// (The legacy `generate_streaming_config_yaml` YAML emitter that +/// shipped the same shape from a `CollectionPlan` was retired in the +/// Option B unification — see [`crate::emit::backend_push`].) /// `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 @@ -1634,7 +1636,20 @@ fn build_gateway_merge_block(mp: &GatewayMergeProcessor) -> Value { /// 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); + // Option B (post-PR-#287): when `agg_type_override` is set, use + // it as the wire `aggregationType` and emit an empty + // `parameters` object — bypasses the sketch_kind → backend type + // mapping for ExactAgg(Sum/Increase/Count) rows the Replanner + // synthesizes for non-sketch (Sum-shaped) workloads. The + // `sketch_kind` / `sketch_params` fields carry sentinel values + // in this case and are not emitted on the wire. + let (aggregation_type, parameters) = match &agg.agg_type_override { + Some(s) => (s.clone(), json!({})), + None => ( + sketch_kind_to_backend_type(&agg.sketch_kind, &agg.sketch_params).to_string(), + sketch_params_to_json(&agg.sketch_params), + ), + }; let aggregation_input = match agg.aggregation_input { AggregationInput::SketchEnvelope => "sketch_envelope", AggregationInput::Raw => "raw", @@ -1651,7 +1666,7 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue { let window_size = clamp_window_secs(Some(agg.window_secs)) .expect("clamp_window_secs preserves Some"); json!({ - "aggregationType": sketch_kind_to_backend_type(&agg.sketch_kind, &agg.sketch_params), + "aggregationType": aggregation_type, "aggregationSubType": "", "metric": agg.metric_name, "labels": { @@ -2030,6 +2045,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }, BackendAggregation { aggregation_id: "agg1".into(), @@ -2040,6 +2056,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }, ], readouts: vec![ @@ -2096,6 +2113,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }, BackendAggregation { aggregation_id: "agg1".into(), @@ -2106,6 +2124,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }, ], readouts: vec![ @@ -2175,6 +2194,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }], readouts: vec![BackendReadout { aggregation_id: "agg0".into(), @@ -2554,6 +2574,7 @@ mod tests { spatial_filter: String::new(), grouping: vec!["zone".into(), "service".into()], aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }], readouts: vec![], }; @@ -2598,6 +2619,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }], readouts: vec![BackendReadout { aggregation_id: "phase_b_agg0".into(), @@ -2644,6 +2666,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }], readouts: vec![], }; @@ -2667,6 +2690,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), aggregation_input: AggregationInput::Raw, + agg_type_override: None, }], readouts: vec![], }; @@ -3932,6 +3956,7 @@ mod tests { spatial_filter: String::new(), grouping: vec!["zone".to_string()], aggregation_input: AggregationInput::SketchEnvelope, + agg_type_override: None, }], readouts: vec![BackendReadout { aggregation_id: "agg0".to_string(), diff --git a/control_plane/src/emit/trait_def.rs b/control_plane/src/emit/trait_def.rs index 977a4227..2302ae10 100644 --- a/control_plane/src/emit/trait_def.rs +++ b/control_plane/src/emit/trait_def.rs @@ -236,6 +236,7 @@ mod tests { grouping: Vec::new(), aggregation_input: crate::physical::colored_dag::emitter::AggregationInput::SketchEnvelope, + agg_type_override: None, }], readouts: vec![BackendReadout { aggregation_id: "agg0".to_string(), diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 08aad275..a80fa473 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -34,7 +34,7 @@ use tracing::{info, warn}; use optimizer::engine::QueryOptimizer; use physical::allocator::SketchAllocator; use pipeline::{Analyzer, QuerySpec}; -use emit::{generate_agent_collector_config, build_precompute_engine_jobs}; +use emit::{generate_agent_collector_config, build_precompute_engine_jobs, post_typed_backend_for_role}; use workload::WorkloadRegistry; use emit::{AgentRuntime, emit_for_runtime}; use types::AgentCollectorConfig; @@ -241,6 +241,16 @@ async fn main() { } } } + // Option B (defensive): re-fire the + // cumulative replan tick on every connect so + // the backend's swap-installed + // streaming-config always reflects every + // planned `(metric, role)` pair. Idempotent + // — cumulative + swap means re-POSTing the + // same plan is a no-op. Guards against + // start-order races where the backend came + // up AFTER the startup replan_all tick fired. + r.replan_all().await; } }); }) @@ -394,6 +404,16 @@ async fn main() { ); } + // ── Shared per-(metric, role) BackendStageConfig cache ─────────────────── + // Built BEFORE the Replanner so it can be wired through + // `with_backend_routing_cache` — every plan-emit cycle + // (handle_plan, replan triggers, startup replan_all tick, OpAMP + // on-connect tick) reads/writes the SAME cumulative state, so the + // data plane's atomic `handle.swap(new_config)` never wipes + // sibling `(metric, role)` aggregations. + let backend_routing_cache: Arc>> = + Arc::new(Mutex::new(HashMap::new())); + // ── Replanner — closes the SP-8 feedback loop ───────────────────────────── let replanner = { let mut r = Replanner::new( @@ -407,6 +427,9 @@ async fn main() { if let Some(client) = backend_client_shared.as_ref() { r = r.with_backend_client(Arc::clone(client)); } + // Option B: share the cumulative cache so replan triggers + // accumulate against the SAME state `handle_plan` writes to. + r = r.with_backend_routing_cache(Arc::clone(&backend_routing_cache)); // Wire the workload registry so the typed-emit path // (`USE_TYPED_STAGE_SPLIT`) can extend its edge stage config // with the same archive-tier metrics the bootstrap GET path @@ -418,6 +441,21 @@ async fn main() { *replanner_cell.write().await = Some(Arc::clone(&replanner)); *registry_cell.write().await = Some(Arc::clone(&workload_registry)); + // ── Startup replan-tick ─────────────────────────────────────────────────── + // Loop through every `(metric, role)` pair the workload-registry + // pre-pop loop populated and POST the typed cumulative + // streaming-config + storage-routing to the backend. Without this, + // queries that never trigger `POST /api/v1/plan` (the smoke + // harness, bootstrap deployments) hit the data plane's static + // startup config (DDSketch only) and `sum by (zone) (…)` returns + // `ExactAgg(Sum) capability not satisfied`. + // + // Runs BEFORE the HTTP server starts accepting requests so the + // first query never lands on a half-warmed backend. Inline + // (not spawned) for the same reason. + info!("startup replan_all: priming cumulative backend state before HTTP server bind"); + replanner.replan_all().await; + let replan_interval = Duration::from_secs( std::env::var("CONTROLLER_REPLAN_INTERVAL_SECS") .ok() @@ -426,8 +464,6 @@ async fn main() { ); let runtime_samples_store = runtime_samples::RuntimeSamplesStore::new(1024); - let backend_routing_cache: Arc>> = - Arc::new(Mutex::new(HashMap::new())); let state = AppState { analyzer: Arc::new(Analyzer::new()), planner, @@ -700,193 +736,20 @@ async fn handle_plan( } agg.grouping = workload.group_by_labels.clone(); } - // B2 cumulative-emit follow-up: update the - // per-(metric, role) cache with THIS - // iteration's `be`, then BOTH the - // streaming-config emit and the - // storage-routing emit below derive their - // payload from the FULL cache. The - // single-iteration `be` is never sent on - // the wire on its own — every post is - // cumulative across all `(metric, role)` - // pairs the control plane has planned. - // - // Why: the data plane's - // `POST /api/v1/streaming-config` handler - // is `handle.swap(new_config)` (an atomic - // full replace) and the storage-routing - // handler is similarly an atomic per-tenant - // swap. Per-iteration posts overwrite - // siblings: - // * streaming-config — drops the prior - // role's `aggregations`, so a metric - // with both DDSketch (Quantile) and - // ExactAgg (Sum) loses one on the - // backend → `sum by (zone) - // (http_requests_total)` returns - // `ExactAgg(Sum) capability not satisfied` - // (the regression that motivates this - // PR — direct follow-up to #283 which - // made the workload/plan stores - // (metric, role)-keyed but left the - // emit path metric-only). - // * storage-routing — drops other - // metrics' entries → default - // `sketch_store` engine → `archive_miss`. - let cumulative_entries: Vec<((String, AggRole), BackendStageConfig)> = { - let mut cache = st.backend_routing_cache.lock().await; - cache.insert((workload.metric_name.clone(), role), be.clone()); - let mut v: Vec<((String, AggRole), BackendStageConfig)> = cache - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(); - // Deterministic ordering so the emitted - // JSON body is reproducible across runs - // and across test invocations. HashMap - // iteration order would otherwise make - // captured-body regression assertions - // flaky. - v.sort_by(|(a_k, _), (b_k, _)| { - a_k.0 - .cmp(&b_k.0) - .then_with(|| a_k.1.as_str().cmp(b_k.1.as_str())) - }); - v - }; - - // Cumulative streaming-config — one - // `BackendStageConfig` whose `aggregations` - // + `readouts` are the concatenation of - // every cache entry's. The data plane's - // swap installs this single - // multi-aggregation config atomically, so - // ALL roles for ALL metrics survive. - let cumulative_be = BackendStageConfig { - aggregations: cumulative_entries - .iter() - .flat_map(|(_, c)| c.aggregations.iter().cloned()) - .collect(), - readouts: cumulative_entries - .iter() - .flat_map(|(_, c)| c.readouts.iter().cloned()) - .collect(), - }; - - // Phase C: post the cumulative typed L5 - // streaming-config JSON to ASAPQuery-backend - // via the shared BackendClient when - // configured. Without a configured endpoint - // this still no-ops silently — same - // fire-and-forget contract as the existing - // Replanner path. - match emit::emit_backend_streaming_config_json(&cumulative_be) { - Ok(json_doc) => { - info!( - stage = "backend", - aggregations = cumulative_be.aggregations.len(), - readouts = cumulative_be.readouts.len(), - cumulative_pairs = cumulative_entries.len(), - "[USE_TYPED_STAGE_SPLIT] posting typed backend JSON" - ); - if let Some(client) = st.backend_client.as_ref() { - let body = json_doc.to_string(); - match client.post_streaming_config_json(body).await { - Ok(()) => info!( - stage = "backend", - endpoint = %client.endpoint(), - "[USE_TYPED_STAGE_SPLIT] typed backend JSON push succeeded" - ), - Err(e) => warn!( - stage = "backend", - endpoint = %client.endpoint(), - error = %e, - "[USE_TYPED_STAGE_SPLIT] typed backend JSON push failed; \ - next replan cycle will retry" - ), - } - } else { - info!( - stage = "backend", - "[USE_TYPED_STAGE_SPLIT] no backend client configured; \ - skipping JSON push (set CONTROLLER_BACKEND_ENDPOINT to enable)" - ); - } - } - Err(e) => warn!(error = %e, "emit_backend_streaming_config_json failed"), - } - - // Phase α (MVP) cumulative storage-routing. - // The routing classifier - // (`build_routing_entry` in - // `emit/stage_config.rs`) reads - // `cfg.aggregations` to derive shape - // routing, so we MUST merge every role's - // aggregations for one metric into a single - // `BackendStageConfig` before passing it - // through — otherwise a metric with both - // DDSketch (Quantile) and ExactAgg (Sum) - // would emit only the last-cached role's - // shape classifications and route the - // siblings to archive. - // - // `emit_backend_storage_routing`'s signature - // is `&[(String, &BackendStageConfig)]` — - // per-metric, NOT per-(metric, role) — so - // the merge happens at the call site (per - // the PR's no-signature-change constraint). - let mut by_metric: std::collections::BTreeMap = - std::collections::BTreeMap::new(); - for ((m, _r), cfg) in &cumulative_entries { - let entry = by_metric.entry(m.clone()).or_insert_with(|| { - BackendStageConfig { - aggregations: Vec::new(), - readouts: Vec::new(), - } - }); - entry.aggregations.extend(cfg.aggregations.iter().cloned()); - entry.readouts.extend(cfg.readouts.iter().cloned()); - } - let routing_owned: Vec<(String, BackendStageConfig)> = - by_metric.into_iter().collect(); - let routing_input: Vec<(String, &BackendStageConfig)> = routing_owned - .iter() - .map(|(k, v)| (k.clone(), v)) - .collect(); - match emit::emit_backend_storage_routing(&routing_input) { - Ok(routing_doc) => { - info!( - stage = "backend", - metric = %workload.metric_name, - cumulative_metrics = routing_owned.len(), - cumulative_pairs = cumulative_entries.len(), - "[USE_TYPED_STAGE_SPLIT] posting cumulative storage-routing JSON" - ); - if let Some(client) = st.backend_client.as_ref() { - let body = routing_doc.to_string(); - match client.post_storage_routing_json(body).await { - Ok(()) => info!( - stage = "backend", - metric = %workload.metric_name, - "[USE_TYPED_STAGE_SPLIT] storage-routing JSON push succeeded" - ), - Err(e) => warn!( - stage = "backend", - metric = %workload.metric_name, - error = %e, - "[USE_TYPED_STAGE_SPLIT] storage-routing JSON push failed; \ - next replan cycle will retry" - ), - } - } else { - info!( - stage = "backend", - "[USE_TYPED_STAGE_SPLIT] no backend client configured; \ - skipping storage-routing JSON push" - ); - } - } - Err(e) => warn!(error = %e, "emit_backend_storage_routing failed"), - } + // Option B unification: every typed cumulative + // emit (handle_plan here, Replanner triggers + // below, startup pre-pop tick, OpAMP + // on-connect tick) flows through the same + // helper. See [`post_typed_backend_for_role`] + // doc for the swap-semantics rationale + + // cumulative-cache contract. + post_typed_backend_for_role( + st.backend_client.as_ref(), + &st.backend_routing_cache, + &workload.metric_name, + role, + be, + ).await; // Mention stage_id so `match` arms aren't // collapsed into untagged log lines if the diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 871392f2..33f1d541 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -365,6 +365,28 @@ pub struct BackendAggregation { /// Phase ε.2 implements the raw-input ingest path. #[serde(default)] pub aggregation_input: AggregationInput, + + /// Option B (post-PR-#287) — when `Some(s)`, the wire-side + /// `aggregationType` is `s` (e.g. `"Sum"`, `"Increase"`, + /// `"MinMax"`) and the `parameters` object is emitted as `{}`, + /// bypassing the sketch-kind → backend-type mapping that runs + /// for the regular sketched aggregations. + /// + /// Why: the typed `bind_workload_typed` rule chain only knows + /// how to lower sketch-shaped statistics (Quantile / Cardinality + /// / Frequency / TopK). Sum/Rate/Count workloads — `sum by + /// (zone) (http_requests_total)`, `rate(metric[5m])`, + /// `count(metric)` — currently decline binding (return None) so + /// the typed L5 stage-split emits nothing for them. Under the + /// Option B unification, the Replanner falls back to this + /// override shape to emit an `ExactAgg(Sum)` (or Increase / + /// Count) row into the cumulative streaming-config so the data + /// plane recognises the metric and `sum by (zone) (…)` queries + /// resolve. `sketch_kind` / `sketch_params` carry sentinel + /// values when the override is in effect (their emitted form is + /// suppressed in `build_backend_aggregation_json`). + #[serde(default)] + pub agg_type_override: Option, } /// Phase ε.1 — what wire shape the backend ingests for an aggregation. @@ -544,6 +566,8 @@ impl Emitter for ThreeStageEmitter { grouping: Vec::new(), // Mode 1 — sketch built at edge, ships envelope. aggregation_input: AggregationInput::SketchEnvelope, + // Regular sketch path — no override. + agg_type_override: None, }); } // Gateway: SketchMerge over edge sketches → one merge @@ -622,6 +646,8 @@ impl Emitter for ThreeStageEmitter { grouping: Vec::new(), // Mode 2 — backend builds sketch from raw OTLP. aggregation_input: AggregationInput::Raw, + // Regular sketch path — no override. + agg_type_override: None, }); } _ => {} diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 5ecf5aa5..b1b3efbd 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -17,19 +17,20 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock}; use tracing::{info, warn}; -use crate::backend_client::{push_or_log, BackendClient}; +use crate::backend_client::BackendClient; use crate::emit::{ build_precompute_engine_jobs, collect_metric_to_family, emit_for_runtime, extend_edge_with_demo_plumbing, generate_agent_collector_config, - generate_streaming_config_yaml, AgentRuntime, WorkloadRegistry, + post_typed_backend_for_role, AgentRuntime, WorkloadRegistry, }; use crate::monitor::Scraper; use crate::opamp::{OpampServer, RemoteConfig}; use crate::optimizer::baseline::BaselinePlanner; use crate::optimizer::{cost as cost_model, rules}; +use crate::physical::colored_dag::emitter::BackendStageConfig; use crate::physical::stage_split; use crate::store::{PlanStore, WorkloadStore}; use crate::types::QueryWorkload; @@ -53,14 +54,27 @@ pub struct Replanner { scraper: Arc, opamp_endpoint: String, /// Optional client for pushing newly-generated `StreamingConfig` - /// YAML to the ASAPQuery-backend's `/api/v1/streaming-config` + /// JSON to the ASAPQuery-backend's `/api/v1/streaming-config` /// endpoint. When present, every successful replan POSTs the new - /// plan to the backend in addition to the existing OpAMP pushes - /// to agent-role and backend-role collectors. Configured via the + /// plan to the backend through [`post_typed_backend_for_role`] — + /// same typed cumulative path the HTTP `POST /api/v1/plan` handler + /// in `main::handle_plan` uses. Configured via the /// `CONTROLLER_BACKEND_ENDPOINT` env var; defaults to `None` so /// existing deployments that don't yet run ASAPQuery-backend /// behave exactly as before. backend_client: Option>, + /// Shared per-`(metric, role)` `BackendStageConfig` cache used by + /// [`post_typed_backend_for_role`]. Holding it on the `Replanner` + /// means SLA-violation / plan-expiry replans, startup pre-pop + /// ticks, and OpAMP on-connect ticks all derive their cumulative + /// POST from the SAME state `handle_plan` writes to — so the + /// data plane's atomic `handle.swap` swap never loses sibling + /// `(metric, role)` aggregations. + /// + /// `None` when no backend is configured (the helper is still + /// invoked — it logs and returns). + backend_routing_cache: + Option>>>, /// Optional handle to the control-plane-wide [`WorkloadRegistry`]. /// Used only by the typed-emit path /// ([`Replanner::try_emit_typed_edge_yaml`]) to extend the edge @@ -96,13 +110,19 @@ impl Replanner { scraper, opamp_endpoint: opamp_endpoint.into(), backend_client: None, + backend_routing_cache: None, workload_registry: None, agent_to_metrics: Arc::new(RwLock::new(HashMap::new())), } } /// Attach a [`BackendClient`] so every replan also pushes the new - /// `StreamingConfig` YAML to the ASAPQuery-backend via HTTP. + /// typed cumulative `StreamingConfig` + `BackendStorageRouting` + /// JSON to the ASAPQuery-backend via HTTP — through the same + /// [`post_typed_backend_for_role`] helper `handle_plan` uses, so + /// the data plane's atomic swap never loses sibling `(metric, + /// role)` aggregations. + /// /// Builder-style — call during control plane startup in `main.rs`. /// Without this call, replans continue to push only via OpAMP and /// the ASAPQuery-backend (if running) keeps its startup config. @@ -111,6 +131,23 @@ impl Replanner { self } + /// Attach the shared per-`(metric, role)` `BackendStageConfig` + /// cache so the typed cumulative emit reads from + writes to the + /// SAME state `main::handle_plan` mutates. Without this the + /// Replanner's cumulative POSTs would derive from an empty + /// cache and overwrite `handle_plan`'s state on every fire. + /// + /// Builder-style; safe to omit (the typed emit still works — it + /// just operates on a Replanner-local cache, which is fine when + /// the Replanner is the sole producer, e.g. in unit tests). + pub fn with_backend_routing_cache( + mut self, + cache: Arc>>, + ) -> Self { + self.backend_routing_cache = Some(cache); + self + } + /// Attach the control-plane-wide [`WorkloadRegistry`] so the typed /// emit path (gated by `USE_TYPED_STAGE_SPLIT`) can extend the /// edge stage config with the workload-registry archive metrics @@ -446,24 +483,44 @@ impl Replanner { self.opamp.push(&agent_id, cfg).await; } } - // Push the ASAPQuery-backend StreamingConfig YAML via HTTP if a - // backend client is configured. This is the producer side of the - // ASAPQuery PR E hot-reload contract: the backend receives the - // new plan on its /api/v1/streaming-config endpoint and makes it - // visible to the next query without restarting. - if let Some(backend_client) = self.backend_client.as_ref() { - match generate_streaming_config_yaml(metric, &plan) { - Ok(yaml) => { - push_or_log(backend_client, metric, yaml).await; - } - Err(e) => { - warn!( - metric, - error = %e, - "failed to build ASAPQuery streaming-config YAML — \ - skipping backend HTTP push for this replan cycle" - ); - } + // Option B: post the typed cumulative `StreamingConfig` + + // `BackendStorageRouting` JSON to the backend through the + // SAME helper `main::handle_plan` uses. The shared + // `backend_routing_cache` (when wired via + // `with_backend_routing_cache`) is updated under the helper's + // lock and the cumulative POST surfaces every `(metric, role)` + // pair the controller has planned — so the data plane's + // atomic `handle.swap(new_config)` never wipes sibling + // aggregations the way the retired + // `generate_streaming_config_yaml` single-aggregation YAML + // path did. + // + // The cache is shared with `AppState`; if the Replanner was + // built without one (test fixture), we fall back to a + // throwaway local cache so the helper still emits — the + // cumulative semantics degrade gracefully (the Replanner is + // the sole writer in that scenario). + if stage_split::typed_stage_split_enabled() { + if let Some(be) = self.build_backend_stage_config(&workload, role) { + let fallback_cache = self.backend_routing_cache.clone(); + let cache_arc = fallback_cache.unwrap_or_else(|| { + Arc::new(Mutex::new(HashMap::new())) + }); + post_typed_backend_for_role( + self.backend_client.as_ref(), + cache_arc.as_ref(), + metric, + role, + be, + ) + .await; + } else { + warn!( + metric, + role = %role, + "could not build BackendStageConfig for replan — \ + skipping backend HTTP push for this cycle" + ); } } @@ -479,6 +536,134 @@ impl Replanner { true } + /// Run the same planner → stage-split → Backend extraction + /// `handle_plan` runs, then return the patched + /// `BackendStageConfig` ready for [`post_typed_backend_for_role`]. + /// + /// Mirrors the L4/L5 flow in `main::handle_plan` for specs that + /// supply only explicit fields (no `query_string`): runs + /// `bind_workload_typed` to lower the workload to a + /// `PhysicalExpr`, then `split_typed_three_stage` to extract the + /// per-stage configs, finds the `Backend` arm, and patches + /// `metric_name` / `window_secs` / `grouping` on each + /// `BackendAggregation` from the workload spec (same patch the + /// `handle_plan` Backend arm applies). + /// + /// **ExactAgg fallback (Option B)**: when `bind_workload_typed` + /// declines (Sum/Rate/Count workloads — `sum by (zone) + /// (http_requests_total)`, `rate(metric[5m])`, + /// `count(metric)`) AND the role classifies as + /// Sum/Count/Other/Topk, synthesize a single ExactAgg-shaped + /// `BackendStageConfig` carrying an `agg_type_override` of + /// `"Sum"` / `"Increase"` / `"MinMax"` so the cumulative + /// streaming-config still surfaces the metric to the backend. + /// Without this fallback the typed cumulative POST would omit + /// every Sum-shaped metric and `sum by (zone) (…)` queries would + /// return `No result for query`. + fn build_backend_stage_config( + &self, + workload: &QueryWorkload, + role: AggRole, + ) -> Option { + // ── Typed sketch path (Quantile / Cardinality / TopK / Frequency) ── + if let Some(physical_expr) = rules::bind_workload_typed(workload) { + if let Some(configs) = stage_split::split_typed_three_stage(&physical_expr) { + if let Some(mut be) = configs.into_iter().find_map(|(_, cfg)| match cfg { + crate::physical::colored_dag::StageConfig::Backend(be) => Some(be), + _ => None, + }) { + // Same patch the `handle_plan` Backend arm applies: the L5 + // emitter leaves `metric_name` empty when path-recovery + // through `extract_edge_facts` fails, and always leaves + // `grouping` empty (`QueryExpr::Aggregate.by` is positional + // `ColumnId`s with no label-name resolution today). The + // `QueryWorkload` carries both unambiguously, and every + // aggregation under one workload shares them. + 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(); + } + return Some(be); + } + } + } + + // ── ExactAgg fallback (Sum / Count / Increase) ──────────────────── + // + // The typed binder declined — most likely because the workload is + // Sum/Rate/Count-shaped (raw passthrough, no sketch family). Emit a + // single-aggregation `BackendStageConfig` with an + // `agg_type_override` so the data plane gets an ExactAgg entry it + // can dispatch to its `SumAccumulator` / `IncreaseAccumulator`. + let agg_type_override = match role { + AggRole::Sum => Some("Sum"), + AggRole::Count => Some("Sum"), // count(metric) maps to a Sum-as-count accumulator on the backend + AggRole::Other => None, + AggRole::Quantile | AggRole::Topk => None, + }; + let agg_type_override = agg_type_override?.to_string(); + use crate::physical::colored_dag::emitter::{ + AggregationInput, BackendAggregation, BackendStageConfig, + }; + use crate::sketch_algebra::params::{DDSketchParams, SketchKind, SketchParams}; + let window_secs = workload.time_window.as_secs().max(1); + Some(BackendStageConfig { + aggregations: vec![BackendAggregation { + aggregation_id: format!("exact-{}-{}", workload.metric_name, role), + metric_name: workload.metric_name.clone(), + // Sentinel sketch_kind / sketch_params — `agg_type_override` + // takes precedence in `build_backend_aggregation_json`, so + // these are not emitted on the wire. DDSketch is the + // chosen sentinel because every backend that recognises + // `AggregationType::FromStr` also accepts DDSketch (and + // we don't have a `SketchKind::None` variant today). + sketch_kind: SketchKind::DDSketch, + sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + window_secs, + spatial_filter: String::new(), + grouping: workload.group_by_labels.clone(), + // ExactAgg consumes raw values at the backend (the agent + // ships counter samples; the backend's + // SumAccumulator integrates them). + aggregation_input: AggregationInput::Raw, + agg_type_override: Some(agg_type_override), + }], + // No readout entries — ExactAgg produces the answer + // directly; the readout dispatch happens at PromQL eval + // time on the backend. + readouts: Vec::new(), + }) + } + + /// Loop through every `(metric, role)` pair in the `WorkloadStore` + /// and call [`Self::replan_metric_role`]. Used at startup (after + /// the workload-registry pre-pop) and on OpAMP first-connect so + /// the backend's cumulative `StreamingConfig` carries every + /// planned `(metric, role)` BEFORE the first query lands — + /// without this, queries that don't trigger `POST /api/v1/plan` + /// hit the data plane's static startup config (DDSketch only) and + /// fail. + /// + /// Idempotent: every call replays the cumulative POST. Re-runs + /// over the same set of pairs are a no-op on the backend (same + /// shape ⇒ same `handle.swap` payload). + pub async fn replan_all(&self) { + let keys = self.workload_store.keys(); + if keys.is_empty() { + info!("replan_all: workload store empty — nothing to plan"); + return; + } + info!(count = keys.len(), "replan_all: planning every (metric, role) pair"); + for (metric, role) in keys { + self.replan_metric_role(&metric, role).await; + } + } + /// Re-plans every `(metric, role)` pair whose `valid_until` has /// already passed. pub async fn replan_expired(&self) {