From 165f4b9751d5a136cc2bdc2129bcca6ec92be17a Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 17:00:14 -0400 Subject: [PATCH 1/3] mvp v6 phase C: add AgentRole::Gateway variant + header parser Phase B left gateway YAML pushes as info!-logged because no AgentRole existed for the mid-tier collector. Phase C extends the OpAMP role vocabulary with `Gateway` so the typed L5 stage_split path can route the gateway YAML directly via push_to_role, identical to how it already routes Agent and Backend. The from_header parser is now public (used by tests) and recognises "gateway" case-insensitively; unknown values still fall back to Agent so legacy / mis-configured collectors keep working unchanged. Tests: gateway round-trips through serde and through a real WebSocket connect path (verified via connected_agents_with_roles). Co-Authored-By: Claude Opus 4.7 (1M context) --- controller/src/opamp/mod.rs | 53 ++++++++++++++++++++++++++++++++++++- 1 file changed, 52 insertions(+), 1 deletion(-) diff --git a/controller/src/opamp/mod.rs b/controller/src/opamp/mod.rs index c7375253..9b797631 100644 --- a/controller/src/opamp/mod.rs +++ b/controller/src/opamp/mod.rs @@ -54,14 +54,24 @@ pub struct AgentStatus { pub enum AgentRole { /// Edge / agent collector that produces sketches. Agent, + /// Mid-tier gateway collector that forwards / merges sketches + /// between edge agents and the backend. Phase C (MVP v6) wires + /// this role through OpAMP so the typed L5 stage_split path can + /// push the gateway YAML directly via `push_to_role`. + Gateway, /// Backend / aggregation collector that merges sketches. Backend, } impl AgentRole { - fn from_header(value: &str) -> Self { + /// Parse the `X-Agent-Role` header value into an `AgentRole`. + /// Recognises `backend`, `gateway`, and `agent` (case-insensitive); + /// any other value (including the empty string) defaults to + /// `Agent` so legacy / mis-configured collectors keep working. + pub fn from_header(value: &str) -> Self { match value.trim().to_lowercase().as_str() { "backend" => AgentRole::Backend, + "gateway" => AgentRole::Gateway, _ => AgentRole::Agent, } } @@ -446,6 +456,47 @@ mod tests { assert_eq!(AgentRole::from_header("BACKEND"), AgentRole::Backend); } + /// Phase C: `Gateway` is a recognised role and round-trips through + /// the OpAMP `X-Agent-Role` header parser. This locks in the wire + /// vocabulary that the typed L5 stage_split path relies on when it + /// calls `push_to_role(AgentRole::Gateway, ...)` and expects to + /// reach gateway-role collectors only. + #[test] + fn role_from_header_recognises_gateway() { + assert_eq!(AgentRole::from_header("gateway"), AgentRole::Gateway); + assert_eq!(AgentRole::from_header("Gateway"), AgentRole::Gateway); + assert_eq!(AgentRole::from_header("GATEWAY"), AgentRole::Gateway); + // Round-trip through serde lowercase rename. + let s = serde_json::to_string(&AgentRole::Gateway).unwrap(); + assert_eq!(s, "\"gateway\""); + let back: AgentRole = serde_json::from_str(&s).unwrap(); + assert_eq!(back, AgentRole::Gateway); + // Still distinct from the other two roles. + assert_ne!(AgentRole::Gateway, AgentRole::Agent); + assert_ne!(AgentRole::Gateway, AgentRole::Backend); + } + + /// Phase C: a gateway-role client connecting via WebSocket appears + /// in `connected_agents_with_roles` tagged as `Gateway`. Together + /// with the from_header test above this proves the role plumbs + /// through the connect path that `push_to_role` selects on. + #[tokio::test] + async fn gateway_role_round_trips_through_connection() { + let (srv, addr) = start_server().await; + let _gateway_ws = connect_ws_client(addr, "gw-1", "gateway").await; + + // Wait for server-side registration to complete. + for _ in 0..20 { + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + let map = srv.connected_agents_with_roles().await; + if map.get("gw-1") == Some(&AgentRole::Gateway) { + return; + } + } + let map = srv.connected_agents_with_roles().await; + panic!("gateway role never registered; map = {:?}", map); + } + /// Helper: start a real OpAMP server on a random port, return the server /// Arc and the bound address. async fn start_server() -> (Arc, std::net::SocketAddr) { From e4bb33397aec3925240c97c9f06054e1ae9f0b31 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 17:02:13 -0400 Subject: [PATCH 2/3] mvp v6 phase C: share BackendClient between AppState and Replanner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase B's typed L5 stage_split path can only info!-log the backend JSON it emits because the BackendClient that knows how to POST to ASAPQuery-backend's /api/v1/streaming-config lives only on the Replanner. Phase C lifts the client out of Replanner-local state into a shared `Option>` built once in main(), passed by clone to both the Replanner (existing path: post a YAML on every replan) and AppState (Phase C path: post the typed L5 backend JSON from handle_plan, wired in the next commit). Picked the Arc-shared-reference approach over moving ownership: the diff is smaller (no Replanner API churn — `with_backend_client` is unchanged) and lets either side push without needing back-references. Tests: AppState's backend_client is None by default (preserves the no-endpoint silently-skip contract) and Some when constructed with a URL — the production path that reads CONTROLLER_BACKEND_ENDPOINT. Co-Authored-By: Claude Opus 4.7 (1M context) --- controller/src/main.rs | 86 +++++++++++++++++++++++++++++++++++------- 1 file changed, 73 insertions(+), 13 deletions(-) diff --git a/controller/src/main.rs b/controller/src/main.rs index ad3e01db..9438d1f2 100644 --- a/controller/src/main.rs +++ b/controller/src/main.rs @@ -67,6 +67,16 @@ struct AppState { /// agents' `sketch-runtime::PushExporter`. Read by decision /// loops in the replanner. runtime_samples: Arc, + /// Phase C (MVP v6): shared `BackendClient` for posting + /// `StreamingConfig` JSON / YAML to the ASAPQuery-backend's + /// `POST /api/v1/streaming-config` endpoint. Phase B had this + /// only on the `Replanner`, so the typed L5 stage_split path in + /// `handle_plan` could only `info!`-log the backend JSON it + /// emitted. Sharing via `Arc` lets `AppState` and `Replanner` + /// both push without owning a duplicate client. `None` when + /// `CONTROLLER_BACKEND_ENDPOINT` is unset, matching the + /// pre-existing fire-and-forget contract. + backend_client: Option>, } // ── Entry point ─────────────────────────────────────────────────────────────── @@ -247,6 +257,28 @@ async fn main() { } } + // ── Phase C: shared BackendClient ───────────────────────────────────────── + // Built once at startup; shared between Replanner (existing path — + // pushes the StreamingConfig YAML on every successful replan) and + // AppState (Phase C — pushes the typed L5 backend JSON emitted by + // `emit_backend_config_json` from `handle_plan`). `None` when + // `CONTROLLER_BACKEND_ENDPOINT` is unset preserves the + // fire-and-forget "skip silently" contract from Phase B. + let backend_client_shared: Option> = + backend_endpoint.as_ref().map(|endpoint| { + info!( + endpoint = %endpoint, + "ASAPQuery-backend StreamingConfig push enabled" + ); + Arc::new(backend_client::BackendClient::new(endpoint.clone())) + }); + if backend_client_shared.is_none() { + info!( + "ASAPQuery-backend StreamingConfig push disabled \ + (set CONTROLLER_BACKEND_ENDPOINT= to enable)" + ); + } + // ── Replanner — closes the SP-8 feedback loop ───────────────────────────── let replanner = { let mut r = Replanner::new( @@ -257,19 +289,8 @@ async fn main() { Arc::clone(&scraper), opamp_ep.clone(), ); - if let Some(endpoint) = backend_endpoint.as_ref() { - info!( - endpoint = %endpoint, - "ASAPQuery-backend StreamingConfig push enabled" - ); - r = r.with_backend_client(Arc::new(backend_client::BackendClient::new( - endpoint.clone(), - ))); - } else { - info!( - "ASAPQuery-backend StreamingConfig push disabled \ - (set CONTROLLER_BACKEND_ENDPOINT= to enable)" - ); + if let Some(client) = backend_client_shared.as_ref() { + r = r.with_backend_client(Arc::clone(client)); } Arc::new(r) }; @@ -297,6 +318,7 @@ async fn main() { opamp_endpoint: opamp_ep, workload_registry: Arc::clone(&workload_registry), runtime_samples: Arc::clone(&runtime_samples_store), + backend_client: backend_client_shared, }; // ── Background tasks ────────────────────────────────────────────────────── @@ -794,6 +816,16 @@ fn short_hash(s: &str) -> String { /// No background tasks are started; OpAMP/scraper hold no real connections. #[cfg(test)] fn test_app() -> (AppState, axum::Router) { + test_app_with_backend(None) +} + +/// Phase C test helper: build an `AppState` whose `backend_client` is +/// optionally set to a real `BackendClient` pointed at a mock URL. The +/// `None` arm is the legacy path used by every existing test; +/// `Some(url)` is the new entry point for Phase C tests that exercise +/// the typed L5 backend-JSON push. +#[cfg(test)] +fn test_app_with_backend(backend_url: Option) -> (AppState, axum::Router) { let online_store = init_online_store(); let plan_store = Arc::new(PlanStore::new()); let workload_store = Arc::new(WorkloadStore::new()); @@ -812,6 +844,8 @@ fn test_app() -> (AppState, axum::Router) { Arc::clone(&scraper), "ws://ctrl:4320/v1/opamp", )); + let backend_client = backend_url + .map(|u| Arc::new(backend_client::BackendClient::new(u))); let state = AppState { analyzer: Arc::new(Analyzer::new()), planner, @@ -824,6 +858,7 @@ fn test_app() -> (AppState, axum::Router) { opamp_endpoint: "ws://ctrl:4320/v1/opamp".into(), workload_registry: Arc::new(WorkloadRegistry::empty()), runtime_samples: runtime_samples::RuntimeSamplesStore::new(64), + backend_client, }; let router = axum::Router::new() .route("/api/v1/plan", axum::routing::post(handle_plan)) @@ -860,6 +895,31 @@ mod api_tests { }) } + // ── Phase C: AppState.backend_client wiring ─────────────────────────────── + + /// Default-constructed AppState (no `CONTROLLER_BACKEND_ENDPOINT`) + /// must leave `backend_client` as `None` so the typed L5 backend + /// JSON push silently no-ops, matching the Phase B fire-and-forget + /// contract. + #[test] + fn app_state_backend_client_none_by_default() { + let (state, _router) = test_app(); + assert!(state.backend_client.is_none(), + "backend_client should default to None when no endpoint is configured"); + } + + /// When constructed with a backend URL (the production path takes + /// it from `CONTROLLER_BACKEND_ENDPOINT`), the field is populated + /// and ready for the Phase C `handle_plan` push. + #[test] + fn app_state_backend_client_some_when_constructed_with_url() { + let (state, _router) = test_app_with_backend( + Some("http://127.0.0.1:1/api/v1/streaming-config".into()), + ); + let bc = state.backend_client.expect("backend_client must be Some"); + assert_eq!(bc.endpoint(), "http://127.0.0.1:1/api/v1/streaming-config"); + } + // ── POST /api/v1/plan ───────────────────────────────────────────────────── #[tokio::test] From b8378495f7668ae3f7fe680d348d644591239753 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 17:04:06 -0400 Subject: [PATCH 3/3] mvp v6 phase C: wire typed L5 gateway YAML push + backend JSON POST MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase B's typed L5 stage_split path produces three per-stage configs but only the edge YAML reached collectors. The Gateway and Backend arms of the StageConfig match in handle_plan were info!-logged because the role / client wiring didn't exist yet. Phase C closes both gaps: 1. StageConfig::Gateway → emit_gateway_yaml + push_to_role(Gateway). Mirrors the existing edge push exactly, now that AgentRole::Gateway is in the OpAMP role vocabulary (commit 1). 2. StageConfig::Backend → emit_backend_config_json + new BackendClient::post_streaming_config_json. The typed L5 emitter produces a serde_json::Value (vs the YAML the existing replanner path pushes), so the new method POSTs application/json with the same 2xx-or-error contract as push_streaming_config. Skips silently when CONTROLLER_BACKEND_ENDPOINT is unset. Existing behaviour is unchanged when USE_TYPED_STAGE_SPLIT is unset — the typed branch still gates on planner::stage_split::typed_stage_split_enabled(). Tests: - backend_client: json_post_round_trips_body + json_post_non_2xx_is_error - opamp: push_to_role_gateway_routes_only_to_gateway_role (verifies routing — does not decode the protobuf, sidestepping a pre-existing decode-tag-zero issue in sibling tests) Co-Authored-By: Claude Opus 4.7 (1M context) --- controller/src/backend_client.rs | 71 ++++++++++++++++++++++++++++++++ controller/src/main.rs | 69 ++++++++++++++++++++++++------- controller/src/opamp/mod.rs | 58 ++++++++++++++++++++++++++ 3 files changed, 182 insertions(+), 16 deletions(-) diff --git a/controller/src/backend_client.rs b/controller/src/backend_client.rs index c79b4a7f..3c563a2b 100644 --- a/controller/src/backend_client.rs +++ b/controller/src/backend_client.rs @@ -94,6 +94,42 @@ impl BackendClient { )) } } + + /// Phase C (MVP v6) variant of [`Self::push_streaming_config`] + /// that POSTs `application/json`. The typed L5 + /// `emit_backend_config_json` emitter produces a `serde_json::Value` + /// rather than a YAML document, and the ASAPQuery-backend's + /// `/api/v1/streaming-config` endpoint accepts both content types + /// (PR #297 / Phase B documents the JSON shape). Same 2xx-or-error + /// contract as the YAML variant; same fire-and-forget semantics + /// at the call site. + pub async fn post_streaming_config_json(&self, json: String) -> Result<()> { + debug!( + endpoint = %self.endpoint, + json_bytes = json.len(), + "posting streaming-config JSON to ASAPQuery-backend" + ); + let resp = self + .http + .post(&self.endpoint) + .header("content-type", "application/json") + .body(json) + .send() + .await + .context("failed to POST streaming-config JSON to backend")?; + + let status = resp.status(); + if status.is_success() { + Ok(()) + } else { + let body = resp.text().await.unwrap_or_default(); + Err(anyhow::anyhow!( + "backend returned {} for streaming-config JSON POST: {}", + status, + body + )) + } + } } /// Fire-and-forget convenience helper used by the replanner. Logs @@ -187,4 +223,39 @@ mod tests { // Must not panic or propagate — fire-and-forget semantics. push_or_log(&client, "cpu_usage", "content".to_string()).await; } + + /// Phase C: the JSON variant POSTs the body verbatim, returns + /// `Ok(())` on a 2xx, and surfaces non-2xx as `Err`. Mock backend + /// captures the body so we can verify it round-trips. + #[tokio::test] + async fn json_post_round_trips_body() { + let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); + 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(); + client + .post_streaming_config_json(json.clone()) + .await + .expect("json post ok"); + + let received = sink.0.lock().unwrap(); + assert_eq!(received.len(), 1); + assert_eq!(received[0], json); + } + + /// Phase C: non-2xx from the backend surfaces as an error so the + /// caller (handle_plan) can log + move on. + #[tokio::test] + async fn json_post_non_2xx_is_error() { + let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); + let url = + start_mock_backend(sink.clone(), axum::http::StatusCode::BAD_REQUEST).await; + + let client = BackendClient::new(url); + let result = client.post_streaming_config_json("{}".to_string()).await; + assert!(result.is_err(), "expected error on 400, got {result:?}"); + let msg = result.unwrap_err().to_string(); + assert!(msg.contains("400"), "error msg should mention 400: {msg}"); + } } diff --git a/controller/src/main.rs b/controller/src/main.rs index 9438d1f2..243fbb97 100644 --- a/controller/src/main.rs +++ b/controller/src/main.rs @@ -484,28 +484,65 @@ async fn handle_plan( } } crate::stage_split::StageConfig::Gateway(gw) => { - // No `AgentRole::Gateway` exists today - // (Phase C adds it). Log the YAML so the - // demo overlay can pick it up via stdout - // until Phase C wires the role. + // Phase C: AgentRole::Gateway is now wired + // through the OpAMP role-routing path, so + // the gateway YAML is pushed to gateway-role + // collectors the same way the edge YAML is + // pushed to agent-role collectors above. match config::emit_gateway_yaml(&gw, &st.opamp_endpoint) { - Ok(yaml) => info!( - stage = "gateway", bytes = yaml.len(), - yaml = %yaml, - "[USE_TYPED_STAGE_SPLIT] gateway YAML emitted (push deferred to Phase C)" - ), + Ok(yaml) => { + let hash = short_hash(&yaml); + info!( + stage = "gateway", bytes = yaml.len(), + "[USE_TYPED_STAGE_SPLIT] pushing typed gateway YAML" + ); + st.opamp.push_to_role( + AgentRole::Gateway, + RemoteConfig { config_hash: hash, yaml }, + ).await; + } Err(e) => warn!(error = %e, "emit_gateway_yaml failed"), } } crate::stage_split::StageConfig::Backend(be) => { + // Phase C: post the 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 config::emit_backend_config_json(&be) { - Ok(json_doc) => info!( - stage = "backend", - aggregations = be.aggregations.len(), - readouts = be.readouts.len(), - json = %json_doc, - "[USE_TYPED_STAGE_SPLIT] backend streaming-config JSON emitted (push deferred to Phase C)" - ), + Ok(json_doc) => { + info!( + stage = "backend", + aggregations = be.aggregations.len(), + readouts = be.readouts.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_config_json failed"), } // Mention stage_id so `match` arms aren't diff --git a/controller/src/opamp/mod.rs b/controller/src/opamp/mod.rs index 9b797631..5b906d75 100644 --- a/controller/src/opamp/mod.rs +++ b/controller/src/opamp/mod.rs @@ -568,6 +568,64 @@ mod tests { assert_eq!(String::from_utf8(rc.config_hash).unwrap(), "hash-1", "delivered hash must match"); } + /// Phase C integration test: gateway YAML emitted from the typed L5 + /// stage_split path is queued onto the gateway-role connection and + /// not onto agent-role / backend-role connections. We don't decode + /// the protobuf (an unrelated decode-tag-zero issue affects sibling + /// tests today); we only assert *delivery routing* — the gateway + /// client receives a non-empty binary frame within the timeout, the + /// other roles receive nothing. + #[tokio::test] + async fn push_to_role_gateway_routes_only_to_gateway_role() { + use futures_util::StreamExt; + let (srv, addr) = start_server().await; + let mut agent_ws = connect_ws_client(addr, "agent-1", "agent").await; + let mut gateway_ws = connect_ws_client(addr, "gateway-1", "gateway").await; + let mut backend_ws = connect_ws_client(addr, "backend-1", "backend").await; + + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + // Mirror the call site that handle_plan now exercises: + // emit_gateway_yaml(...) → push_to_role(Gateway, ...). + // We use a stand-in YAML payload here; the emitter has its own + // tests in stage_config.rs. + let yaml = "extensions:\n opamp: {}\n".to_string(); + srv.push_to_role(AgentRole::Gateway, RemoteConfig { + config_hash: "hash-gw".into(), + yaml, + }).await; + + // Gateway must receive exactly one frame. + let msg = tokio::time::timeout( + std::time::Duration::from_secs(2), + gateway_ws.next(), + ) + .await + .expect("gateway timed out") + .unwrap() + .unwrap(); + let bytes = msg.into_data(); + assert!(!bytes.is_empty(), "gateway must receive a non-empty frame"); + + // Other roles must receive nothing within a short window. + let agent_result = tokio::time::timeout( + std::time::Duration::from_millis(200), + agent_ws.next(), + ).await; + assert!( + agent_result.is_err(), + "agent-role client must not receive gateway-role push" + ); + let backend_result = tokio::time::timeout( + std::time::Duration::from_millis(200), + backend_ws.next(), + ).await; + assert!( + backend_result.is_err(), + "backend-role client must not receive gateway-role push" + ); + } + /// `push_to_role(Agent)` must not deliver to a backend-role client. #[tokio::test] async fn push_to_agent_role_does_not_reach_backend_role() {