From 1adf6fbde0a5b73bb176407787e21dab6ac98ac6 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 10:30:09 -0600 Subject: [PATCH 1/4] docs: replace historical commentary with current contracts --- control_plane/src/asap_tier_implement.rs | 19 +- control_plane/src/backend_client.rs | 37 +-- control_plane/src/emit/backend_push.rs | 49 +--- control_plane/src/emit/mod.rs | 48 +--- control_plane/src/emit/otap.rs | 23 +- control_plane/src/emit/stage_config.rs | 223 ++++-------------- control_plane/src/emit/telegraf.rs | 10 +- control_plane/src/main.rs | 117 ++------- control_plane/src/opamp/mod.rs | 4 +- .../src/physical/colored_dag/allocator.rs | 6 +- control_plane/src/physical/colored_dag/dag.rs | 38 +-- .../src/physical/colored_dag/emitter.rs | 57 ++--- control_plane/src/physical/colored_dag/mod.rs | 41 +--- .../src/physical/colored_dag/stage_id.rs | 53 +---- .../src/physical/deployment_cost/mod.rs | 3 - .../deployment_cost/sketch_capability.rs | 13 +- .../src/physical/deployment_cost/wire.rs | 58 +---- .../src/physical/post_asap/cost_model.rs | 38 +-- .../src/physical/post_asap/deployment_expr.rs | 41 +--- control_plane/src/physical/post_asap/lower.rs | 44 +--- .../src/physical/post_asap/matcher.rs | 7 +- control_plane/src/physical/post_asap/tests.rs | 48 +--- .../src/physical/runtime_capability.rs | 67 +----- control_plane/src/physical/stage_split.rs | 14 +- .../src/physical/workload_planner.rs | 37 +-- control_plane/src/query_parser/mod.rs | 36 +-- control_plane/src/replan.rs | 20 +- control_plane/src/types.rs | 15 +- control_plane/src/types_v2.rs | 18 +- control_plane/src/workload.rs | 44 +--- crates/asap_types/src/aggregation_config.rs | 13 +- crates/asap_types/src/aggregation_type.rs | 22 +- crates/asap_types/src/enums.rs | 22 +- crates/asap_types/src/policy_fingerprint.rs | 9 +- data_plane/src/drivers/ingest/otel.rs | 43 +--- .../src/drivers/ingest/series_resolver.rs | 16 +- data_plane/src/main.rs | 134 ++--------- data_plane/src/monitor/coordinator.rs | 35 +-- data_plane/src/monitor/mod.rs | 36 +-- .../precompute_engine/accumulator_factory.rs | 22 +- data_plane/src/precompute_engine/config.rs | 9 +- data_plane/src/precompute_engine/engine.rs | 10 +- .../src/precompute_engine/ingest_handler.rs | 36 +-- .../operators/count_sketch_accumulator.rs | 32 +-- .../src/precompute_engine/output_sink.rs | 17 +- data_plane/src/precompute_engine/worker.rs | 17 +- .../query_engines/asap_query_engine/engine.rs | 114 ++------- .../asap_query_engine/live_serve.rs | 36 +-- .../query_engines/asap_query_engine/mod.rs | 14 +- .../asap_query_engine/post_asap_readout.rs | 51 +--- .../asap_query_engine/summary_exec.rs | 32 +-- .../asap_query_engine/summary_executor.rs | 40 +--- .../routing/backend_storage_routing.rs | 124 +--------- .../routing/capability_matching.rs | 4 +- .../routing/query_engine_routing.rs | 20 +- .../thanos_query_engine/forward.rs | 2 +- .../src/storage_engines/sketch_db/accuracy.rs | 9 +- .../storage_engines/sketch_db/backfill/mod.rs | 61 +---- .../sketch_db/backfill/processor.rs | 108 ++------- .../sketch_db/backfill/prometheus_reader.rs | 2 +- .../sketch_db/backfill/service.rs | 16 +- .../sketch_db/backfill/window_builder.rs | 51 +--- .../sketch_db/backfill/worker.rs | 68 +----- .../src/storage_engines/sketch_db/data/mod.rs | 26 +- .../sketch_db/index/epoch_columnar.rs | 39 +-- .../storage_engines/sketch_db/index/mod.rs | 65 ++--- .../sketch_db/lifecycle/eviction.rs | 65 +---- .../sketch_db/lifecycle/mod.rs | 24 +- .../sketch_db/lifecycle/reconcile.rs | 35 +-- .../src/storage_engines/sketch_db/mod.rs | 32 +-- .../sketch_db/query/asap_tier_result.rs | 17 +- .../storage_engines/sketch_db/query/mod.rs | 43 +--- .../sketch_db/query/timeline.rs | 44 +--- data_plane/src/storage_engines/traits.rs | 5 +- .../types/precomputed_output.rs | 13 +- .../storage_engines/types/streaming_config.rs | 7 +- data_plane/src/tests/mod.rs | 7 - .../tests/test_utilities/engine_factories.rs | 18 +- data_plane/src/tests/test_utilities/mod.rs | 8 +- 79 files changed, 490 insertions(+), 2411 deletions(-) diff --git a/control_plane/src/asap_tier_implement.rs b/control_plane/src/asap_tier_implement.rs index 02809866a..48853ab83 100644 --- a/control_plane/src/asap_tier_implement.rs +++ b/control_plane/src/asap_tier_implement.rs @@ -213,12 +213,7 @@ mod tests { #[test] fn bare_selector_has_no_aggregate_root_to_implement() { - // L1 adoption (design-target-architecture.md Part B), accepted - // behavior change -- see - // asap_tier_analysis::bare_selector_is_no_longer_asap_tier_answerable's - // comment: `lower_promql` doesn't implicitly wrap a bare selector - // in `Aggregate { Sum }` the way the retired local parser did, so - // there's no `Aggregate` node here at all to find a root at. + // A bare selector has no `Aggregate` node, so there is no ASAP-tier root. let roots = implement_promql_for_asap_tier("http_requests_total").expect("parses and implements"); assert!(roots.is_empty(), "{roots:?}"); @@ -324,16 +319,8 @@ mod tests { /// behavior shift. #[test] fn implement_frequency_as_agg_test() { - // Per this test's own prior instructions: the gap it used to - // document (under-realizing to `Logical` because `asap-plan` had - // no `Extension`/`Frequency` opinion) is now closed -- not via an - // `Extension` hook, but because L1 adoption - // (design-target-architecture.md Part B) makes `count_over_time` - // lower directly to `AggIntent::Count { accuracy: Epsilon(...) }` - // (a real, first-class, non-exact intent) rather than needing - // this deployment's `Frequency` extension wrapper at all -- - // `asap-plan` realizes a non-exact `Count` as a real CMS-backed - // `SummaryAgg` + `SummaryEstimate` on its own. + // Approximate `count_over_time` lowers to a first-class `Count` intent, + // which Planner realizes as CMS-backed `SummaryAgg` + `SummaryEstimate`. let roots = implement_promql_for_asap_tier("count_over_time(http_requests_total[5m])") .expect("parses and implements"); assert_eq!(roots.len(), 1); diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 75841913c..13e2179ca 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -167,14 +167,8 @@ impl BackendClient { } } - /// Phase C (MVP v6) variant of [`Self::push_streaming_config`] - /// that POSTs `application/json`. The typed L5 - /// `emit_backend_streaming_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. + /// Post streaming configuration as `application/json`. The backend accepts + /// both JSON and YAML; a non-2xx response is an error for the caller to log. pub async fn post_streaming_config_json(&self, json: String) -> Result<()> { debug!( endpoint = %self.endpoint, @@ -280,22 +274,9 @@ impl BackendClient { } } - /// Phase α (MVP) sibling of [`Self::post_streaming_config_json`]: - /// POSTs the control-plane-emitted `BackendStorageRouting` JSON - /// document to the backend's `POST /api/v1/storage_routing` - /// endpoint. The backend hot-loads the routing table and the next - /// instant query consults the new table. - /// - /// Endpoint resolution: the field [`Self::endpoint`] is the - /// control plane's configured streaming-config endpoint (e.g. - /// `http://backend.svc:8088/api/v1/streaming-config`). We rewrite - /// the path component from `/api/v1/streaming-config` to - /// `/api/v1/storage_routing` so operators only configure one - /// `CONTROL_PLANE_BACKEND_ENDPOINT` env var and both pushes land - /// at the same backend host. URLs that don't end in - /// `/api/v1/streaming-config` are passed through unchanged - /// (a test-mode escape hatch — the unit test below builds a - /// mock URL ending in `/storage_routing` directly). + /// Post backend storage-routing JSON. Derive the URL by replacing the + /// `/api/v1/streaming-config` suffix with `/api/v1/storage_routing`; URLs + /// without that suffix are used verbatim. pub async fn post_storage_routing_json(&self, json: String) -> Result<()> { let url = derive_storage_routing_url(&self.endpoint); debug!( @@ -533,7 +514,7 @@ mod tests { push_or_log(&client, "cpu_usage", "content".to_string()).await; } - /// Phase C: the JSON variant POSTs the body verbatim, returns + /// 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] @@ -553,7 +534,7 @@ mod tests { assert_eq!(received[0], json); } - /// Phase C: non-2xx from the backend surfaces as an error so the + /// 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() { @@ -567,7 +548,7 @@ mod tests { assert!(msg.contains("400"), "error msg should mention 400: {msg}"); } - /// Phase α: storage-routing-URL derivation rewrites the path + /// storage-routing-URL derivation rewrites the path /// component when the configured endpoint ends in /// `/api/v1/streaming-config`, leaving everything else untouched. #[test] @@ -648,7 +629,7 @@ mod tests { server.abort(); } - /// Phase α: full happy path. A mock backend hosts the storage + /// full happy path. A mock backend hosts the storage /// routing endpoint; the client POSTs the control-plane-emitted JSON /// and the body round-trips verbatim. Mirrors `json_post_round_trips_body`. async fn start_mock_routing_backend( diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index 9be93c761..5efc89971 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -1,42 +1,13 @@ -//! Typed cumulative push of `BackendStageConfig` to the ASAPQuery-backend. +//! Cumulative backend configuration push shared by planning and replanning. //! -//! 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: +//! Each push updates the per-`(metric, role)` cache, concatenates all aggregations +//! and readouts in deterministic order, then posts streaming configuration. +//! Storage routing merges those entries by metric before posting. Both endpoints +//! replace their configuration atomically, so a push must preserve sibling roles +//! and metrics. //! -//! 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. +//! Emission and transport errors log at WARN and return. The next planning cycle +//! retries with the latest configuration. use std::collections::{BTreeMap, HashMap}; // `Future` is only referenced by the now-test-only `retry_transient` @@ -1005,7 +976,7 @@ mod tests { /// re-plan, so the restarted backend recovers its config. #[tokio::test] async fn repost_after_simulated_backend_reset_re_pushes_full_config() { - // Phase 1: initial plan lands on the first backend instance. + // initial plan lands on the first backend instance. let (url1, mock1) = start_dual_mock(axum::http::StatusCode::OK, axum::http::StatusCode::OK).await; let client1 = StdArc::new(BackendClient::new(url1)); @@ -1022,7 +993,7 @@ mod tests { assert_eq!(mock1.streaming_hits.load(StdOrdering::SeqCst), 1); assert_eq!(mock1.routing_hits.load(StdOrdering::SeqCst), 1); - // Phase 2: the backend silently restarts — model it as a brand-new + // the backend silently restarts — model it as a brand-new // mock with zero recorded hits. NOTHING expires, NO replan fires. let (url2, mock2) = start_dual_mock(axum::http::StatusCode::OK, axum::http::StatusCode::OK).await; diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index 8a7e1db49..edbcf3791 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -1,19 +1,5 @@ -//! `emit/` — per-deployment-model plan emitters (L5 output side). -//! -//! Per `control_plane/docs/design.md` §5 `core::emit`. The 2026-05 -//! layered-cleanup refactor consolidated the former -//! `controller/src/config/` directory here. Mapping: -//! -//! | Old path | New path | -//! |---|---| -//! | `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` | *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/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`] | -//! | `config/stage_config_telegraf.rs` | [`telegraf`] | -//! | `config/workloads.rs` | [`crate::workload`] (top-level — design.md §5 puts `workload` next to `emit`, not inside it) | +//! Per-deployment plan emitters. Typed stage configs become collector +//! configuration, backend streaming configuration, and storage routing. pub mod agent; pub mod backend_push; @@ -50,25 +36,9 @@ use anyhow::Result; use planner_types::post_asap::{SketchAlgorithm, SummaryExpr, SummaryNode}; use std::rc::Rc; -/// Phase ε.1.5 — which edge runtime an agent identifies as. -/// -/// Today every agent the controller has built for runs the OTel-collector -/// (`AsapOtel`); Phase ε.1.5 adds the two new runtime variants the -/// per-runtime emitters target. The runtime is reported by the agent on -/// OpAMP `on_connect` (header `X-Agent-Runtime`); when absent (legacy -/// agents) the controller defaults to `AsapOtel` so the existing -/// behaviour is preserved. -/// -/// Phase ε.1.5 commits the enum + emit-dispatch function. Threading the -/// runtime through OpAMP `on_connect` and into the typed L5 emit path -/// is a follow-up — the emitters can be exercised in isolation today -/// (the Phase ε.1.5 test suite does exactly that). -/// -/// Naming history: the variants were originally `Sketchcollector` / -/// `Sketchotap` / `Sketchtelegraf`; the rename to `AsapOtel` / -/// `AsapOtap` / `AsapTelegraf` (PR `refactor/rename-edge-runtimes-...`) -/// drops the v0 `sketch*` prefix in favour of the symmetric `asap-*` -/// namespace. `from_header` accepts both forms during transition. +/// Edge runtime reported through the OpAMP `X-Agent-Runtime` header. +/// Missing headers default to `AsapOtel`. `from_header` accepts both +/// `asap-*` and `sketch*` names for compatibility. #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "kebab-case")] #[derive(Default)] @@ -97,7 +67,7 @@ impl AgentRuntime { } } -/// Phase ε.1.5 — dispatch the edge emit by agent runtime. Mirrors +/// dispatch the edge emit by agent runtime. Mirrors /// `emit_edge_yaml`'s `(cfg, opamp_endpoint) -> String` shape; the OTAP /// and Telegraf emitters take an additional optional Prometheus URL /// override which we pass through `prometheus_url`. @@ -300,11 +270,7 @@ fn extract_from_plan(plan: &PostAsapPlan) -> Option { fn extract_from_node(node: &Rc) -> Option { match &node.expr { - // `SummaryAgg`'s `kind`/`params` collapsed into one `family: - // SummaryFamilyType` field (ASAPPlanner#218 -- see - // control_plane/docs/design-asapplanner-pin-migration.md); the - // exact-vs-sketch check this used to need `is_exact_accumulator` - // for is now which enum variant `family` is. + // The `family` variant distinguishes exact accumulators from sketches. SummaryExpr::SummaryAgg { family: planner_types::post_asap::SummaryFamilyType::Sketch(kind, _), .. diff --git a/control_plane/src/emit/otap.rs b/control_plane/src/emit/otap.rs index 6d377c011..b26d14354 100644 --- a/control_plane/src/emit/otap.rs +++ b/control_plane/src/emit/otap.rs @@ -1,4 +1,4 @@ -//! Phase ε.1.5 — OTAP Dataflow DAG YAML emitter (per-runtime mirror of +//! OTAP Dataflow DAG YAML emitter (per-runtime mirror of //! [`super::stage_config::emit_edge_yaml`]). //! //! `asap-otap` uses the otap-dataflow Rust runtime; its config surface is @@ -9,10 +9,7 @@ //! and registers under the URN `urn:otel:exporter:otlp_http` via //! `linkme`'s `distributed_slice(OTAP_EXPORTER_FACTORIES)`. //! -//! Three modes (the placement decision itself is made upstream in -//! `stage_split`; the `BindMode`-shaped selector Phase ε.1 sketched out -//! in `physical::deployment_cost::wire` was never wired in and was removed in the -//! 2026-07 retirement pass — see that module's doc): +//! Placement modes selected by the upstream stage splitter: //! //! 1. `SketchAtEdge` — DAG includes a sketch processor node between the //! OTLP receiver and the OTLP gRPC exporter to the gateway. (The @@ -26,7 +23,7 @@ //! (`/api/v1/otlp/v1/metrics`). //! //! The function consumes the same [`EdgeStageConfig`] the OTel-collector -//! emitter does — Phase ε.1.5 keeps the typed L5 plan as the single +//! emitter does, keeping the typed plan as the single //! source of truth across all three runtimes. The emitter dispatches per //! `EdgeStageConfig` via the same `prometheus_archive_metrics` / //! `sketch_processors` signals the OTel-collector emitter uses (Mode 3 @@ -60,19 +57,11 @@ const URN_OTLP_GRPC_EXPORTER: &str = "exporter:otlp_grpc"; /// URN of the OTLP receiver (gRPC + HTTP). const URN_OTLP_RECEIVER: &str = "receiver:otlp"; -/// URN of the asap_sketches processor registered in the otap-patch tree. -/// Phase ε.1.5 wires the URN abstractly; the binary side lands the -/// plugin in `otap-patch/plugins/asap_sketches/` per -/// [`docs/design-asap-otap-rust-integration.md`]. +/// URN of the sketch processor registered by the OTAP plugin. const URN_ASAP_SKETCHES_PROCESSOR: &str = "processor:asap_sketches"; -// ── DAG YAML structural types ──────────────────────────────────────────────── -// -// These mirror the otap-dataflow `engine`/`groups`/`pipelines`/`nodes` -// schema in `otel-arrow/rust/otap-dataflow/configs/*.yaml`. We keep a -// minimal set to round-trip; the full schema (channel_capacity policies, -// engine settings, etc.) is left at struct defaults — Phase ε.1.5 only -// commits the wiring shape, not policy. +// Structural subset of the OTAP DAG configuration. Omitted engine and channel +// policy fields retain their runtime defaults. #[derive(Serialize)] struct OtapDag { diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 6d94e1643..a4ab229e8 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -1,50 +1,12 @@ -//! Phase B (MVP v6) — turn a typed L5 [`StageConfig`] map (produced by -//! [`crate::physical::colored_dag::ThreeStageEmitter`]) into the **wire bytes** the -//! three executors actually consume: +//! Turn typed stage configs into wire configuration for edge, gateway, and backend. //! -//! - [`emit_edge_yaml`] → OTel-collector YAML for the edge agent (OTLP -//! receiver → per-sketch processor(s) → OTLP exporter to gateway). -//! - [`emit_gateway_yaml`] → OTel-collector YAML for the gateway -//! 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, -//! 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 -//! the per-metric sketch families from the typed [`BackendStageConfig`] -//! inputs and turns them into `(metric, [target])` rows the backend's -//! HTTP query handler consults via `BackendStorageRouting::lookup_with_shape`. +//! * [`emit_edge_yaml`] and [`emit_gateway_yaml`] produce collector YAML. +//! * [`emit_backend_streaming_config_json`] produces precompute configuration. +//! * [`emit_backend_storage_routing`] maps metrics and query shapes to engines. //! -//! 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 `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. The `opamp_endpoint` -//! parameter is the controller's WebSocket URL the emitted YAML's -//! `extensions.opamp` block must point at; the caller threads it -//! through from `AppState::opamp_endpoint`. The `agent_id` parameter -//! is the identity the agent presents in the `X-Agent-ID` WS header -//! when it reconnects after a controller-pushed restart (Issue #2 — -//! without this header the controller's OpAMP server can't re-identify -//! the agent). Broadcast callers that don't have a single agent in -//! scope pass the literal placeholder `"$AGENT_ID"` and rely on the -//! agent container's env to expand it at boot. -//! -//! NOTE: the memory_limiter soft threshold the 5-sketch routing path -//! emits can be tuned via the controller's `ASAP_AGENT_MEMORY_LIMIT_MIB` -//! env var (default 1280 MiB). Operators bumping the agent container's -//! cgroup limit raise both together. See `emit_edge_yaml_5sketch_routing`. +//! These are pure transformations. Callers supply the OpAMP endpoint and agent +//! identity; broadcast configs use `$AGENT_ID` for expansion by each collector. +//! Backend pushes go through the cumulative helper in `backend_push`. use anyhow::{Context, Result}; use serde::Serialize; @@ -179,26 +141,9 @@ pub fn clamp_window_secs(w: Option) -> Option { w.map(|s| s.clamp(MIN_WINDOW_SECS, MAX_WINDOW_SECS)) } -/// Process-level gate selecting the FUSED single-pipeline `asap_edge` -/// edge wire shape (issue #46) over the legacy `routing`-connector -/// per-family fan-out. -/// -/// **Default OFF** so the established 5-sketch routing emit (and its -/// large unit-test surface) is unchanged for callers who haven't -/// migrated the agent build yet. Set `ASAP_EDGE_FUSED=1` (or -/// `true` / `yes`) on the controller process to switch every -/// `metric_to_family`-populated edge config to the fused -/// `[memory_limiter, cumulativetodelta, asap_edge]` pipeline that the -/// new fused agent processor consumes. -/// -/// We gate on an env var (mirroring `typed_stage_split_enabled()` / -/// `ASAP_AGENT_MEMORY_LIMIT_MIB`) rather than a new `EdgeStageConfig` -/// field so the change is additive: no struct-literal churn across the -/// ~17 construction sites, no serde wire-shape bump, and the two emit -/// paths read the IDENTICAL `cfg` fields. The flag is the canonical -/// migration switch — once the fused agent build is the default -/// deployment the gate's default flips to ON (and the routing path is -/// retired). +/// Process-level switch for the fused `asap_edge` pipeline. Set +/// `ASAP_EDGE_FUSED=1` (or `true` / `yes`) to enable it. Default is off. +/// Both fused and per-family routing consume the same `EdgeStageConfig`. pub fn fused_asap_edge_enabled() -> bool { matches!( std::env::var("ASAP_EDGE_FUSED").as_deref(), @@ -208,18 +153,9 @@ pub fn fused_asap_edge_enabled() -> bool { // ── Public API ──────────────────────────────────────────────────────────────── -/// Build the OTel-collector YAML for an edge agent from the typed L5 -/// [`EdgeStageConfig`] payload. -/// -/// The `opamp_endpoint` is embedded under `extensions.opamp.server.ws.endpoint` -/// so the agent can receive runtime config updates without restart. -/// -/// The emitter does NOT resolve `ExportTarget::Stage(_)` to a concrete -/// network address; Phase C plumbs a `DeploymentConstraints` resolver -/// that maps the symbolic stage role to e.g. `gateway:4317`. Until -/// then, we emit a documented placeholder (`gateway:4317`) so the YAML -/// is syntactically valid and round-trips through Otel's loader for -/// integration tests. +/// Build collector YAML from a typed [`EdgeStageConfig`]. The OpAMP endpoint +/// lets the agent receive configuration updates. Symbolic export targets use +/// default hostnames unless an explicit endpoint is supplied. pub fn emit_edge_yaml( cfg: &EdgeStageConfig, opamp_endpoint: &str, @@ -388,7 +324,7 @@ pub fn emit_edge_yaml( // ── Phase ε.1 Mode 3 / Phase 3.2.5 Bug (b) — per-pipeline routing ───── // Two routing axes can fire from a single edge agent: // - // * Phase ε.1: Mode-3 metrics carry `asap.mode = prometheus_archive` + // * Mode-3 metrics carry `asap.mode = prometheus_archive` // as a data-point attribute and dispatch to the Prometheus OTLP // receiver via a separate `otlphttp/prometheus` exporter. // * Phase 3.2.5 Bug (b): warm-passthrough metrics (the freshness @@ -650,25 +586,10 @@ pub fn emit_gateway_yaml( serde_yaml::to_string(&doc).context("serialize gateway stage config") } -/// Build the JSON document the ASAPQuery-backend's -/// `POST /api/v1/streaming-config` endpoint accepts, sourced from the -/// typed L5 [`BackendStageConfig`]. +/// Build backend streaming-config JSON from [`BackendStageConfig`]. /// -/// 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 -/// query engine can prepare per-readout dispatch entries up-front (the -/// existing YAML form has no readouts list because the legacy planner -/// materialises one aggregation per metric and infers readouts from the -/// PromQL query at execution time; Phase B's typed `BackendStageConfig` -/// carries the readouts explicitly, so we ship them too — backends that -/// don't recognise the field will ignore it without erroring). +/// Emit aggregation rows and explicit readouts. Omit `aggregationId`: the backend +/// derives policy identity from configuration content. pub fn emit_backend_streaming_config_json( cfg: &BackendStageConfig, monitors: &[crate::emit::monitor::MonitorIntent], @@ -704,7 +625,7 @@ pub fn emit_backend_streaming_config_json( Ok(doc) } -/// Phase α (MVP): build the JSON document the ASAPQuery-backend's +/// build the JSON document the ASAPQuery-backend's /// `POST /api/v1/storage_routing` endpoint accepts, sourced from the /// typed L5 [`BackendStageConfig`] payloads emitted by [`crate::planner::stage_split`]. /// @@ -736,7 +657,7 @@ pub fn emit_backend_streaming_config_json( /// } /// ``` /// -/// ## Classification rules (Phase α) +/// ## Classification rules /// /// For each `(metric, BackendStageConfig)` we derive a target list by /// inspecting the L4 sketch families landed at the backend: @@ -805,7 +726,7 @@ pub fn emit_backend_storage_routing_for_tenant( })) } -/// Phase ε.1 — same as [`emit_backend_storage_routing`] but also +/// same as [`emit_backend_storage_routing`] but also /// emits `thanos_query` engine entries for Mode 3 metrics. /// /// Mode-3 metrics have NO `BackendStageConfig` entry (the backend doesn't @@ -902,10 +823,7 @@ fn build_routing_entry(metric_name: &str, cfg: &BackendStageConfig) -> JsonValue if has_hll { warm_shapes.push("count"); } - // Heap-bearing kinds count too — `SummaryKind` (unlike the retired - // `physical::post_asap::SummaryKind`) promotes `with_heap` to a distinct - // identity variant, but a topk-bound Count-Sketch/CMS aggregation - // still needs to register here exactly as it did before the split. + // Heap-bearing frequency sketches also contribute their routing capability. let has_count_sketch = algorithms.iter().any(|k| { matches!( k, @@ -1330,38 +1248,13 @@ fn emit_edge_yaml_5sketch_routing( } } - // ── ASAPCollector#403: edge-aggregate Sum-role counters ──────────────── + // Sum-role counters with grouping labels and no sketch-family assignment + // use a per-metric sum pipeline. This ships one series per grouping tuple. + // Ungrouped counters remain raw passthrough; sketched counters keep their + // sketch pipeline. // - // A Sum-role metric (in `cumulative_counter_metrics`) whose marquee - // queries are `sum by () (...)` previously fell through the - // routing connector's `default_pipelines` into `raw_passthrough` (no - // aggregation): every counter datapoint across the full wire-attr - // cardinality (e.g. ~10k zone×rack×node×pod series) streamed - // continuously to the backend, which did the Sum-by-grouping fan-in - // centrally. That inverts the edge-aggregation value prop and was the - // dominant driver of the asap arm's backend-ingress blowup - // (~12 Mbps of ~12.5 Mbps measured). - // - // Fix (mirrors the static agent config's `metrics/sum_aggregate` - // pipeline): for each Sum-role metric that (a) has grouping_labels - // declared and (b) is NOT routed to any sketch family, register a - // dedicated `metricstransform/sumby_` processor + - // `metrics/sum_aggregate_` pipeline and route the metric - // there instead of letting it default to raw_passthrough. The agent - // then ships one summed series per grouping-label tuple per flush - // window. The backend's `evaluate_exact_agg` produces the identical - // `sum by ()` and per-group `rate` answers at reduced - // cardinality. - // - // gorillas3 (cold-tier archive) still writes RAW full-cardinality - // samples on this pipeline BEFORE the metricstransform collapses the - // stream, preserving cold-fallback drill-down (e.g. - // `count(metric{