diff --git a/control_plane/src/emit/backend.rs b/control_plane/src/emit/backend.rs deleted file mode 100644 index 1d12efe2..00000000 --- a/control_plane/src/emit/backend.rs +++ /dev/null @@ -1,85 +0,0 @@ -use crate::types::*; -use anyhow::Context; -use serde_json::json; - -/// Generate the OTel collector YAML for the **backend-role collector** — -/// the central collector that merges per-agent partial sketches. -/// -/// The pipeline carries a single `{sketch_type}_merge` processor grouped -/// by `cfg.group_by`. (An earlier `_staged` variant added an optional -/// `dedup` processor driven by the legacy `StagedPlan`; that path was -/// retired with the legacy L5 — re-modelling dedup on the typed L5's -/// `BackendStageConfig` is a follow-up if it proves needed.) -pub fn generate_backend_collector_config( - cfg: &BackendCollectorConfig, - opamp_endpoint: &str, -) -> anyhow::Result { - let merge_key = format!("{}_merge", cfg.merge_sketch_type); - - let mut processors = serde_json::Map::new(); - processors.insert( - merge_key.clone(), - json!({ "mode": "merge", "group_by": cfg.group_by }), - ); - - let doc = serde_yaml::to_value(&json!({ - "extensions": { - "opamp": { "server": { "ws": { "endpoint": opamp_endpoint } } } - }, - "processors": processors, - "service": { - "extensions": ["opamp"], - "pipelines": { - "metrics": { "processors": [&merge_key] } - } - } - })) - .context("build backend collector doc")?; - - serde_yaml::to_string(&doc).context("serialize backend collector config") -} - -// ── Tests ───────────────────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn contains_merge_key() { - let cfg = BackendCollectorConfig { - merge_sketch_type: SketchType::DDSketch, - group_by: vec!["host.name".into()], - }; - let yaml = generate_backend_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap(); - assert!( - yaml.contains("ddsketch_merge"), - "YAML should contain merge key\n{yaml}" - ); - assert!( - yaml.contains("host.name"), - "YAML should contain group_by\n{yaml}" - ); - } - - #[test] - fn hll_merge_key() { - let cfg = BackendCollectorConfig { - merge_sketch_type: SketchType::HLL, - group_by: vec![], - }; - let yaml = generate_backend_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap(); - assert!(yaml.contains("HLL_merge"), "{yaml}"); - } - - #[test] - fn contains_opamp_endpoint() { - let ep = "ws://custom-ctrl:9000/v1/opamp"; - let cfg = BackendCollectorConfig { - merge_sketch_type: SketchType::KLL, - group_by: vec![], - }; - let yaml = generate_backend_collector_config(&cfg, ep).unwrap(); - assert!(yaml.contains(ep), "YAML should contain endpoint\n{yaml}"); - } -} diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index d90696be..58caf9ff 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -7,7 +7,7 @@ //! | Old path | New path | //! |---|---| //! | `config/agent.rs` | [`agent`] | -//! | `config/backend.rs` | [`backend`] | +//! | `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/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) | @@ -17,7 +17,6 @@ pub mod agent; pub mod asapquery_backend; -pub mod backend; pub mod otap; pub mod precompute; pub mod stage_config; @@ -26,7 +25,6 @@ pub mod trait_def; pub use agent::generate_agent_collector_config; pub use asapquery_backend::generate_streaming_config_yaml; -pub use backend::generate_backend_collector_config; 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 9f539367..32afabaa 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -22,10 +22,12 @@ //! These four functions are deliberately **stage-shaped**, not //! plan-shaped: the typed L5 emitter has already split the PhysicalExpr //! across edge / gateway / backend, so each function only sees the slice -//! that's relevant to its executor. The legacy emitters in -//! [`crate::config::agent`] / [`crate::config::backend`] still operate -//! on the legacy `AgentCollectorConfig` / `BackendCollectorConfig` — -//! Phase C will gate-flip the demo overlay onto these typed emitters. +//! that's relevant to its executor. The legacy `agent.rs` emitter still +//! operates on the flat `AgentCollectorConfig`; the legacy backend +//! emitter targeted a "backend-role" OTel merge collector tier that was +//! never deployed and has been retired — typed L5 routes `StageId::Backend` +//! directly to asapquery-backend's precompute engine over HTTP via +//! `emit_backend_streaming_config_json`. //! //! All three are pure transformations: no I/O, no env lookup. The //! `opamp_endpoint` parameter is the controller's WebSocket URL the diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index d921c0d3..838fc70b 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, generate_backend_collector_config, build_precompute_engine_jobs}; +use emit::{generate_agent_collector_config, build_precompute_engine_jobs}; use workload::WorkloadRegistry; use emit::{AgentRuntime, emit_for_runtime}; use types::AgentCollectorConfig; @@ -430,7 +430,6 @@ async fn main() { .route("/api/v1/agents", get(handle_agents)) .route("/api/v1/config/:metric", get(handle_get_config)) .route("/api/v1/collector-config/agent", get(handle_bootstrap_agent_config)) - .route("/api/v1/collector-config/backend", get(handle_bootstrap_backend_config)) .route("/api/v1/cost-model", get(handle_cost_model)) .route("/api/v1/tco", post(handle_tco)) .with_state(state) @@ -504,20 +503,6 @@ async fn handle_plan( ).await; } - // ── Push backend config to backend-role collectors ──────────────────────── - // The typed L5 (`split_typed_three_stage`, below) owns the rich - // per-stage backend config now; this legacy push emits the flat - // backend YAML from the SP-3 `plan.backend_config`. - if let Ok(backend_yaml) = generate_backend_collector_config( - &plan.backend_config, &st.opamp_endpoint, - ) { - let hash = short_hash(&backend_yaml); - st.opamp.push_to_role( - AgentRole::Backend, - RemoteConfig { config_hash: hash, yaml: backend_yaml }, - ).await; - } - // ── Phase B (MVP v6): typed L5 stage_split → per-stage emitter ──────────── // Behind the `USE_TYPED_STAGE_SPLIT` env-var gate so existing // control plane behaviour is unchanged unless explicitly opted in. @@ -822,11 +807,6 @@ async fn handle_rollback( config_hash: short_hash(&yaml), yaml, }).await; } - if let Ok(yaml) = generate_backend_collector_config(&plan.backend_config, &st.opamp_endpoint) { - st.opamp.push_to_role(AgentRole::Backend, RemoteConfig { - config_hash: short_hash(&yaml), yaml, - }).await; - } (StatusCode::OK, Json(json!({ "metric": metric, "rolled_back": true }))).into_response() } Err(e) => (StatusCode::BAD_REQUEST, e.to_string()).into_response(), @@ -1115,24 +1095,6 @@ async fn emit_bootstrap_typed( .with_context(|| format!("emit_for_runtime failed for `{metric}`")) } -/// Bootstrap YAML config for backend (merge) collectors. -async fn handle_bootstrap_backend_config( - State(st): State, -) -> impl IntoResponse { - let cfg = types::BackendCollectorConfig { - merge_sketch_type: types::SketchType::DDSketch, - group_by: vec![], - }; - match generate_backend_collector_config(&cfg, &st.opamp_endpoint) { - Ok(yaml) => ( - StatusCode::OK, - [("content-type", "application/yaml")], - yaml, - ).into_response(), - Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), - } -} - /// Returns the diff between the current and previous plan for `metric`. /// 404 if the metric has no plan, 200 with `null` data if no previous plan exists. async fn handle_plan_diff( diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 35fd01c2..e77e911a 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -23,11 +23,11 @@ use tracing::{info, warn}; use crate::backend_client::{push_or_log, 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_backend_collector_config, + extend_edge_with_demo_plumbing, generate_agent_collector_config, generate_streaming_config_yaml, AgentRuntime, WorkloadRegistry, }; use crate::monitor::Scraper; -use crate::opamp::{AgentRole, OpampServer, RemoteConfig}; +use crate::opamp::{OpampServer, RemoteConfig}; use crate::optimizer::baseline::BaselinePlanner; use crate::optimizer::{cost as cost_model, rules}; use crate::physical::stage_split; @@ -357,18 +357,6 @@ impl Replanner { self.opamp.push(&agent_id, cfg.clone()).await; } } - if let Ok(yaml) = generate_backend_collector_config(&plan.backend_config, &self.opamp_endpoint) { - self.opamp - .push_to_role( - AgentRole::Backend, - RemoteConfig { - config_hash: short_hash(&yaml), - yaml, - }, - ) - .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