From 5e0055e4210be1482deea19bee1e10b17eab5c34 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 23 May 2026 13:18:06 -0600 Subject: [PATCH] control-plane: thread cold ship_endpoint + external_labels into EdgeStageConfig PR #311's emit_edge_yaml_asap_edge derived the fused asap_edge cold tier's ship_endpoint from the OTLP backend host (http://:9098/ingest/gorilla) and read external_labels.cluster from ASAP_CLUSTER inline, because EdgeStageConfig carried neither field. Both were wrong: the cold tier ships to the gorilla-merger over HTTP ingest port 10908 (gRPC 10907), not backend:9098. Add cold_ship_endpoint: Option and cold_external_labels: Vec<(String,String)> to EdgeStageConfig and read them in the emitter instead of the inline placeholder. The colored_dag L5 layer is deployment-independent (no DeploymentConstraints plumbed in), so it cannot resolve a real per-deploy endpoint; it populates the single named defaults (default_cold_ship_endpoint / default_cold_external_labels), which a deploy-aware layer can override post-emit the same way exporter_target is. All struct-literal construction sites updated. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/emit/mod.rs | 8 ++ control_plane/src/emit/otap.rs | 6 + control_plane/src/emit/stage_config.rs | 104 +++++++++++------- control_plane/src/emit/telegraf.rs | 8 ++ control_plane/src/emit/trait_def.rs | 2 + .../src/physical/colored_dag/emitter.rs | 57 ++++++++++ 6 files changed, 147 insertions(+), 38 deletions(-) diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index e07b82ba..78833eec 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -482,6 +482,8 @@ mod runtime_tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), }; let collector = emit_for_runtime( @@ -517,6 +519,8 @@ mod runtime_tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), }; let yaml = emit_for_runtime( AgentRuntime::AsapOtap, @@ -550,6 +554,8 @@ mod runtime_tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), }; let toml = emit_for_runtime( AgentRuntime::AsapTelegraf, @@ -883,6 +889,8 @@ mod runtime_tests { )]), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), }; edge_cfg.metric_to_grouping_labels = collect_metric_to_grouping_labels(®istry, &store); diff --git a/control_plane/src/emit/otap.rs b/control_plane/src/emit/otap.rs index 5d62da02..d32e573a 100644 --- a/control_plane/src/emit/otap.rs +++ b/control_plane/src/emit/otap.rs @@ -396,6 +396,8 @@ mod tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -412,6 +414,8 @@ mod tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -432,6 +436,8 @@ mod tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 8c670985..fda215a9 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -53,9 +53,9 @@ use serde_yaml::{Mapping, Value}; use std::collections::{BTreeMap, HashMap}; use crate::physical::colored_dag::emitter::{ - AggregationInput, ArchiveTierMetric, BackendAggregation, BackendReadout, BackendStageConfig, - EdgeSketchProcessor, EdgeStageConfig, ExportTarget, GatewayMergeProcessor, GatewayStageConfig, - PrometheusArchiveMetric, + default_cold_external_labels, default_cold_ship_endpoint, AggregationInput, ArchiveTierMetric, + BackendAggregation, BackendReadout, BackendStageConfig, EdgeSketchProcessor, EdgeStageConfig, + ExportTarget, GatewayMergeProcessor, GatewayStageConfig, PrometheusArchiveMetric, }; use crate::physical::colored_dag::stage_id::StageId; use crate::sketch_algebra::params::{SketchKind, SketchParams}; @@ -1786,22 +1786,19 @@ fn emit_edge_yaml_asap_edge( // cold tier ON whenever the plan declared any archive-tier metric // (the same `archive_tier_metrics` signal that drove the `gorillas3` // processor on the routing path). `block_duration` / `reorder_grace` - // size from the archive window; `external_labels.cluster` comes from - // `ASAP_TENANT`-adjacent deploy config. + // size from the archive window. // - // GAP — REPORTED, NOT FABRICATED: `EdgeStageConfig` does NOT carry a - // cold `ship_endpoint` or an `external_labels` map today (only the - // S3/MinIO knobs `build_gorillas3_yaml` reads from the controller's - // env). The fused agent ships Gorilla blocks to a backend gorilla - // ingest endpoint that does not yet exist (see the Track-2 note in - // the hand-written config). We therefore follow the emitter's - // documented-placeholder convention for unresolved deploy targets: - // derive `ship_endpoint` from the OTLP backend host (analogous to - // how `resolve_export_endpoint` synthesises `backend:4317` / - // `gateway:4317`) and read `cluster` from `ASAP_CLUSTER` (default - // `asap-mvp`). Threading a real per-deploy cold endpoint + - // external-label map needs a new `EdgeStageConfig` field populated - // from the plan — left as a follow-up. + // PR #311 follow-up: `ship_endpoint` and `external_labels` now come + // from the threaded `EdgeStageConfig` cold fields rather than a + // derived placeholder. PR #311 lacked these fields and guessed + // `http://:9098/ingest/gorilla` from the OTLP exporter host + // — WRONG host AND port. The cold tier actually ships to the + // gorilla-merger over HTTP ingest port 10908 (gRPC 10907). When a + // construction site leaves the fields unset (`cold_ship_endpoint: + // None` / empty `cold_external_labels`) we fall back to the single + // named defaults (`default_cold_ship_endpoint` / + // `default_cold_external_labels`) so the emitted endpoint is always + // the correct merger target, never the old backend:9098 guess. let cold_enabled = !cfg.archive_tier_metrics.is_empty(); let cold_block: Value = { // Cold window: smallest declared archive window, else the @@ -1812,17 +1809,20 @@ fn emit_edge_yaml_asap_edge( .filter_map(|m| m.window_secs) .min() .unwrap_or(window_secs); - // ship_endpoint placeholder: the backend's gorilla ingest. We - // reuse the OTLP backend host so the placeholder tracks the - // exporter target (e.g. `backend` → `http://backend:9098/...`). - let backend_host = match &cfg.exporter_target { - ExportTarget::Endpoint(s) => { - s.split(':').next().unwrap_or("backend").to_string() - } - _ => "backend".to_string(), + // ship_endpoint: the threaded per-deploy cold ingest URL (the + // gorilla-merger). Falls back to the named default when the + // plan didn't carry one. + let ship_endpoint = cfg + .cold_ship_endpoint + .clone() + .unwrap_or_else(default_cold_ship_endpoint); + // external_labels: the threaded label tuples; named default + // (`cluster=`) when none were supplied. + let external_labels = if cfg.cold_external_labels.is_empty() { + default_cold_external_labels() + } else { + cfg.cold_external_labels.clone() }; - let ship_endpoint = format!("http://{backend_host}:9098/ingest/gorilla"); - let cluster = std::env::var("ASAP_CLUSTER").unwrap_or_else(|_| "asap-mvp".to_string()); let mut m = Mapping::new(); m.insert("enabled".into(), Value::Bool(cold_enabled)); m.insert("ship_endpoint".into(), Value::String(ship_endpoint)); @@ -1832,7 +1832,9 @@ fn emit_edge_yaml_asap_edge( ); m.insert("reorder_grace".into(), Value::String("2s".to_string())); let mut ext = Mapping::new(); - ext.insert("cluster".into(), Value::String(cluster)); + for (k, v) in external_labels { + ext.insert(Value::String(k), Value::String(v)); + } m.insert("external_labels".into(), Value::Mapping(ext)); Value::Mapping(m) }; @@ -2471,6 +2473,8 @@ mod tests { metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -3443,6 +3447,8 @@ mod tests { metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), }; let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); @@ -3783,6 +3789,8 @@ mod tests { metric_to_family, metric_to_grouping_labels: HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -4069,6 +4077,8 @@ mod tests { metric_to_family, metric_to_grouping_labels: HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -5197,6 +5207,14 @@ mod tests { "unique_users_per_min".into(), "top_endpoint_qps".into(), ], + // PR #311 follow-up: thread the real per-deploy cold ingest + // (the gorilla-merger HTTP ingest on 10908, NOT backend:9098) + // + an explicit external label so the emit test asserts the + // threaded value flows through rather than the named default. + cold_ship_endpoint: Some( + "http://gorilla-merger:10908/ingest/gorilla".into(), + ), + cold_external_labels: vec![("cluster".into(), "asap-mvp".into())], } } @@ -5329,29 +5347,39 @@ mod tests { assert_eq!(cms.get("rows").and_then(|v| v.as_u64()), Some(5)); assert_eq!(cms.get("cols").and_then(|v| v.as_u64()), Some(2048)); - // 5. cold: block present + enabled. + // 5. cold: block present + enabled. The ship_endpoint and + // external label come from the THREADED `EdgeStageConfig` cold + // fields (PR #311 follow-up), NOT a derived placeholder: the cfg + // sets `cold_ship_endpoint = http://gorilla-merger:10908/...` + // and `cold_external_labels = [(cluster, asap-mvp)]`, and the + // emitter must surface exactly those — proving the threading, + // and proving we no longer emit the wrong `backend:9098` guess. let cold = asap_edge.get("cold").expect("cold block present"); assert_eq!( cold.get("enabled").and_then(|v| v.as_bool()), Some(true), "cold.enabled\n{yaml}" ); + assert_eq!( + cold.get("ship_endpoint").and_then(|v| v.as_str()), + Some("http://gorilla-merger:10908/ingest/gorilla"), + "cold.ship_endpoint must be the threaded gorilla-merger ingest (10908), \ + not the old backend:9098 placeholder\n{yaml}" + ); assert!( - cold.get("ship_endpoint") - .and_then(|v| v.as_str()) - .map(|s| s.contains("ingest/gorilla")) - .unwrap_or(false), - "cold.ship_endpoint placeholder\n{yaml}" + !yaml.contains(":9098"), + "must not emit the wrong backend:9098 cold endpoint\n{yaml}" ); assert_eq!( cold.get("block_duration").and_then(|v| v.as_str()), Some("60s") ); - assert!( + assert_eq!( cold.get("external_labels") .and_then(|v| v.get("cluster")) - .is_some(), - "cold.external_labels.cluster\n{yaml}" + .and_then(|v| v.as_str()), + Some("asap-mvp"), + "cold.external_labels.cluster must be the threaded value\n{yaml}" ); // 6. cumulativetodelta lists the Sum-role counters (strict). diff --git a/control_plane/src/emit/telegraf.rs b/control_plane/src/emit/telegraf.rs index b3ff11bc..01a17cf9 100644 --- a/control_plane/src/emit/telegraf.rs +++ b/control_plane/src/emit/telegraf.rs @@ -317,6 +317,8 @@ mod tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -333,6 +335,8 @@ mod tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -353,6 +357,8 @@ mod tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } @@ -505,6 +511,8 @@ mod tests { metric_to_family: std::collections::HashMap::new(), metric_to_grouping_labels: std::collections::HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), }; let toml = emit_telegraf_toml(&cfg, None).expect("emit ok"); assert!(toml.contains("k = 200"), "k not propagated\n{toml}"); diff --git a/control_plane/src/emit/trait_def.rs b/control_plane/src/emit/trait_def.rs index b50bdd85..32a1cf24 100644 --- a/control_plane/src/emit/trait_def.rs +++ b/control_plane/src/emit/trait_def.rs @@ -210,6 +210,8 @@ mod tests { metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: None, + cold_external_labels: Vec::new(), } } diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index cf6d23d6..92af97f9 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -241,6 +241,55 @@ pub struct EdgeStageConfig { /// metric (e.g. quantile-only workloads). #[serde(default, skip_serializing_if = "Vec::is_empty")] pub cumulative_counter_metrics: Vec, + /// PR #311 follow-up — the cold-tier (Gorilla archive) ingest URL the + /// fused `asap_edge` processor ships per-emit Gorilla blocks to. This + /// is the gorilla-merger's HTTP ingest endpoint + /// (`http://gorilla-merger:10908/ingest/gorilla`; the gRPC side is + /// 10907) — NOT the OTLP backend host/port. PR #311 lacked this field + /// and derived a wrong placeholder (`http://:9098/...`) from + /// `exporter_target`; threading the real value here fixes that. + /// + /// `None` ⇒ the emitter falls back to [`default_cold_ship_endpoint`] + /// (a single named default), so legacy / test construction sites that + /// don't populate it still emit a correct merger endpoint. The + /// `colored_dag` L5 layer cannot resolve a real per-deploy endpoint + /// (it is deployment-independent — no `DeploymentConstraints` is + /// plumbed in), so it populates the named default; a future layer that + /// holds deploy info can set a concrete value. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cold_ship_endpoint: Option, + /// PR #311 follow-up — external labels stamped on every cold-tier + /// Gorilla block the fused `asap_edge` processor ships (the merger + /// uses these for cross-cluster disambiguation). Carried as + /// `(label, value)` tuples (deterministic order at the emit site). + /// PR #311 derived `cluster` from the `ASAP_CLUSTER` env inline; this + /// field threads it explicitly. Empty ⇒ the emitter falls back to + /// [`default_cold_external_labels`] (a single named default that reads + /// `ASAP_CLUSTER`, defaulting to `asap-mvp`). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub cold_external_labels: Vec<(String, String)>, +} + +/// Named default for [`EdgeStageConfig::cold_ship_endpoint`]. +/// +/// The cold tier ships per-emit Gorilla blocks to the **gorilla-merger** +/// over HTTP ingest port **10908** (the gRPC ingest side is 10907). This +/// is the one canonical place that default lives — construction sites and +/// the `asap_edge` emitter both route through here rather than inlining +/// the host/port. PR #311's `http://:9098/ingest/gorilla` guess +/// was wrong (wrong host, wrong port); this is the correct merger target. +pub fn default_cold_ship_endpoint() -> String { + "http://gorilla-merger:10908/ingest/gorilla".to_string() +} + +/// Named default for [`EdgeStageConfig::cold_external_labels`]. +/// +/// One `cluster` label, read from `ASAP_CLUSTER` (default `asap-mvp`). +/// Single source of truth for the cold external-label default so the +/// emitter and any construction site agree. +pub fn default_cold_external_labels() -> Vec<(String, String)> { + let cluster = std::env::var("ASAP_CLUSTER").unwrap_or_else(|_| "asap-mvp".to_string()); + vec![("cluster".to_string(), cluster)] } /// Phase 3.2.5 — one archive-tier metric the agent should land in @@ -516,6 +565,14 @@ impl Emitter for ThreeStageEmitter { metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), cumulative_counter_metrics: Vec::new(), + // The colored-DAG layer is deployment-independent (no + // `DeploymentConstraints` is plumbed in here — see the + // module header), so we cannot resolve a real per-deploy cold + // endpoint at this layer. Populate the single named defaults; + // a layer that holds deploy info can overwrite `edge.cold_*` + // post-emit (same pattern as `exporter_target`). + cold_ship_endpoint: Some(default_cold_ship_endpoint()), + cold_external_labels: default_cold_external_labels(), }; let mut backend_aggregations: Vec = Vec::new(); let mut gateway_processors: Vec = Vec::new();