From f782adde9bfdc118c9c46c9b382422f45b7f7aa9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 25 May 2026 17:42:05 -0600 Subject: [PATCH] feat(control_plane): emit cold-archive intchunk format to the edge agent Add a cold-archive format knob so the control plane can tell the edge agent to ship the lossless intchunk cold-part format instead of the default gorilla-XOR fragments. Adds `cold_format` (enum: fragment default | intchunk) and `cold_coldpart_endpoint` to `EdgeStageConfig`. The fused `asap_edge` emit now writes `format: intchunk` + `coldpart_endpoint:` into the agent `cold:` block only when the format is intchunk; the default fragment path emits neither key, leaving the cold block byte-identical to today. Operators opt in per deployment via the `ASAP_COLD_FORMAT=intchunk` env knob (mirrors how `ASAP_CLUSTER` drives the cold external-label default), read in the shared bootstrap/replan stitch. The coldpart endpoint is derived from the fragment ship endpoint (same merger host:port, `/ingest/coldpart` path) unless overridden. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/emit/mod.rs | 116 ++++++++++++++ control_plane/src/emit/otap.rs | 6 + control_plane/src/emit/stage_config.rs | 147 +++++++++++++++++- control_plane/src/emit/telegraf.rs | 8 + control_plane/src/emit/trait_def.rs | 2 + .../src/physical/colored_dag/emitter.rs | 83 ++++++++++ 6 files changed, 358 insertions(+), 4 deletions(-) diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index de9512a5..3c9684f8 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -207,6 +207,49 @@ pub fn extend_edge_with_demo_plumbing( }); } } + + // 4. Cold-archive format opt-in. The colored-DAG L5 layer is + // deployment-independent and can only populate the named default + // (`Fragment`); this bootstrap/replan-scope helper is the first + // place that holds deploy info (env), so it reads the operator's + // `ASAP_COLD_FORMAT` knob (mirrors how `default_cold_external_labels` + // reads `ASAP_CLUSTER`). `intchunk` ⇒ ship the lossless intchunk + // cold-part format; anything else (incl. unset / `fragment`) leaves + // the default gorilla-XOR fragment emit byte-identical. + apply_cold_format_from_env(edge_cfg); +} + +/// Read the `ASAP_COLD_FORMAT` env knob and, when it is `intchunk`, flip +/// `edge_cfg.cold_format` to [`ColdFormat::Intchunk`] and derive the +/// `cold_coldpart_endpoint` from the cold ship endpoint (swapping the path +/// to `/ingest/coldpart`) unless an explicit `ASAP_COLD_COLDPART_ENDPOINT` +/// is supplied. +/// +/// Any value other than `intchunk` (including unset, empty, or `fragment`) +/// is a no-op — the default gorilla-XOR fragment emit stays byte-identical, +/// so there is NO behavior change unless an operator deliberately opts in. +fn apply_cold_format_from_env(edge_cfg: &mut EdgeStageConfig) { + use crate::physical::colored_dag::emitter::{ + coldpart_endpoint_from_ship, default_cold_ship_endpoint, ColdFormat, + }; + let fmt = std::env::var("ASAP_COLD_FORMAT").unwrap_or_default(); + if !fmt.eq_ignore_ascii_case("intchunk") { + return; + } + edge_cfg.cold_format = ColdFormat::Intchunk; + // An explicit endpoint override wins; otherwise derive from the cold + // ship endpoint (same merger host:port, `/ingest/coldpart` path). + if let Ok(ep) = std::env::var("ASAP_COLD_COLDPART_ENDPOINT") { + if !ep.trim().is_empty() { + edge_cfg.cold_coldpart_endpoint = Some(ep); + return; + } + } + let ship = edge_cfg + .cold_ship_endpoint + .clone() + .unwrap_or_else(default_cold_ship_endpoint); + edge_cfg.cold_coldpart_endpoint = Some(coldpart_endpoint_from_ship(&ship)); } // ── MVP §46: planner ↔ 5-sketch emitter stitching ────────────────────────────── @@ -514,6 +557,71 @@ mod runtime_tests { assert_eq!(AgentRuntime::from_header("garbage"), AgentRuntime::AsapOtel); } + #[test] + fn cold_format_env_knob_opts_into_intchunk_and_derives_endpoint() { + // The operator-facing SET path: `ASAP_COLD_FORMAT=intchunk` flips + // the cold format to intchunk and derives the coldpart endpoint + // from the cold ship endpoint (same merger host:port, + // `/ingest/coldpart` path). Unset / `fragment` is a no-op. + use crate::physical::colored_dag::emitter::{ + default_cold_ship_endpoint, ColdFormat, + }; + + fn fixture() -> EdgeStageConfig { + EdgeStageConfig { + source_metric: None, + label_filters: Vec::new(), + window_secs: None, + sketch_processors: Vec::new(), + exporter_target: + crate::physical::colored_dag::emitter::ExportTarget::Stage( + crate::physical::colored_dag::stage_id::StageId::Backend, + ), + prometheus_archive_metrics: Vec::new(), + archive_tier_metrics: Vec::new(), + warm_passthrough_metrics: Vec::new(), + metric_to_family: std::collections::HashMap::new(), + metric_to_grouping_labels: std::collections::HashMap::new(), + cumulative_counter_metrics: Vec::new(), + cold_ship_endpoint: Some(default_cold_ship_endpoint()), + cold_external_labels: Vec::new(), + metric_to_sample_p: std::collections::HashMap::new(), + cold_format: ColdFormat::default(), + cold_coldpart_endpoint: None, + } + } + + // Unset ⇒ no-op (default fragment, no derived endpoint). + { + let _env = crate::test_support::EnvVarGuard::unset("ASAP_COLD_FORMAT"); + let mut cfg = fixture(); + apply_cold_format_from_env(&mut cfg); + assert_eq!(cfg.cold_format, ColdFormat::Fragment); + assert!(cfg.cold_coldpart_endpoint.is_none()); + } + + // `fragment` ⇒ no-op too. + { + let _env = crate::test_support::EnvVarGuard::set("ASAP_COLD_FORMAT", "fragment"); + let mut cfg = fixture(); + apply_cold_format_from_env(&mut cfg); + assert_eq!(cfg.cold_format, ColdFormat::Fragment); + assert!(cfg.cold_coldpart_endpoint.is_none()); + } + + // `intchunk` ⇒ flip + derive coldpart endpoint from ship endpoint. + { + let _env = crate::test_support::EnvVarGuard::set("ASAP_COLD_FORMAT", "intchunk"); + let mut cfg = fixture(); + apply_cold_format_from_env(&mut cfg); + assert_eq!(cfg.cold_format, ColdFormat::Intchunk); + assert_eq!( + cfg.cold_coldpart_endpoint.as_deref(), + Some("http://gorilla-merger:10908/ingest/coldpart"), + ); + } + } + #[test] fn emit_for_runtime_default_matches_emit_edge_yaml() { // Serialize against the env-mutating tests in `stage_config`: @@ -545,6 +653,8 @@ mod runtime_tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, }; let collector = emit_for_runtime( @@ -583,6 +693,8 @@ mod runtime_tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, }; let yaml = emit_for_runtime( AgentRuntime::AsapOtap, @@ -619,6 +731,8 @@ mod runtime_tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, }; let toml = emit_for_runtime( AgentRuntime::AsapTelegraf, @@ -958,6 +1072,8 @@ mod runtime_tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, }; 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 e5a3213c..bc7870ff 100644 --- a/control_plane/src/emit/otap.rs +++ b/control_plane/src/emit/otap.rs @@ -399,6 +399,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -418,6 +420,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -441,6 +445,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 109d2f37..358783ba 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -53,9 +53,10 @@ use serde_yaml::{Mapping, Value}; use std::collections::{BTreeMap, HashMap}; use crate::physical::colored_dag::emitter::{ - default_cold_external_labels, default_cold_ship_endpoint, AggregationInput, ArchiveTierMetric, - BackendAggregation, BackendReadout, BackendStageConfig, EdgeSketchProcessor, EdgeStageConfig, - ExportTarget, GatewayMergeProcessor, GatewayStageConfig, PrometheusArchiveMetric, + coldpart_endpoint_from_ship, default_cold_external_labels, default_cold_ship_endpoint, + AggregationInput, ArchiveTierMetric, BackendAggregation, BackendReadout, BackendStageConfig, + ColdFormat, EdgeSketchProcessor, EdgeStageConfig, ExportTarget, GatewayMergeProcessor, + GatewayStageConfig, PrometheusArchiveMetric, }; use crate::physical::colored_dag::stage_id::StageId; use crate::sketch_algebra::params::{SketchKind, SketchParams}; @@ -1903,7 +1904,30 @@ fn emit_edge_yaml_asap_edge( }; let mut m = Mapping::new(); m.insert("enabled".into(), Value::Bool(cold_enabled)); - m.insert("ship_endpoint".into(), Value::String(ship_endpoint)); + m.insert( + "ship_endpoint".into(), + Value::String(ship_endpoint.clone()), + ); + // Cold-archive format: when the deploy opted into the lossless + // intchunk cold-part format, emit `format: intchunk` + the + // `coldpart_endpoint` so the agent ships to `/ingest/coldpart` + // rather than the default gorilla-XOR fragments. `Fragment` (the + // default) emits NEITHER key, leaving the cold block byte-identical + // to the pre-format emit (`ship_endpoint` only). + if cfg.cold_format == ColdFormat::Intchunk { + m.insert("format".into(), Value::String("intchunk".to_string())); + // coldpart_endpoint: the threaded value, else derived from the + // fragment ship_endpoint by swapping the path to + // `/ingest/coldpart` (same merger host:port). + let coldpart_endpoint = cfg + .cold_coldpart_endpoint + .clone() + .unwrap_or_else(|| coldpart_endpoint_from_ship(&ship_endpoint)); + m.insert( + "coldpart_endpoint".into(), + Value::String(coldpart_endpoint), + ); + } m.insert( "block_duration".into(), Value::String(format!("{block_secs}s")), @@ -2573,6 +2597,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -3555,6 +3581,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, }; let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); @@ -3903,6 +3931,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -4253,6 +4283,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -5414,6 +5446,8 @@ mod tests { ), cold_external_labels: vec![("cluster".into(), "asap-mvp".into())], metric_to_sample_p: HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -5638,6 +5672,109 @@ mod tests { assert!(yaml.contains("otlp/backend:"), "{yaml}"); } + #[test] + fn cold_format_default_fragment_emits_no_format_keys() { + // Default cold_format (Fragment) must NOT emit `format:` or + // `coldpart_endpoint:` in the agent `cold:` block — the cold block + // stays byte-identical to the pre-format emit (ship_endpoint only), + // so there is NO behavior change when the operator leaves the knob + // unset. The default fixture builds with ColdFormat::default(). + let _env = crate::test_support::EnvVarGuard::set("ASAP_EDGE_FUSED", "1"); + let cfg = fused_asap_edge_cfg(); + let yaml = emit_edge_yaml(&cfg, "ws://controller:4320/v1/opamp", "agent-1") + .expect("emit fused asap_edge ok"); + + let doc: serde_yaml::Value = + serde_yaml::from_str(&yaml).unwrap_or_else(|e| panic!("emitted YAML must parse: {e}\n{yaml}")); + let cold = doc + .get("processors") + .and_then(|p| p.get("asap_edge")) + .and_then(|p| p.get("cold")) + .expect("cold block present"); + assert!( + cold.get("format").is_none(), + "default (fragment) cold block must NOT carry a `format:` key\n{yaml}" + ); + assert!( + cold.get("coldpart_endpoint").is_none(), + "default (fragment) cold block must NOT carry a `coldpart_endpoint:` key\n{yaml}" + ); + // The fragment ship_endpoint is unchanged. + assert_eq!( + cold.get("ship_endpoint").and_then(|v| v.as_str()), + Some("http://gorilla-merger:10908/ingest/gorilla"), + "fragment ship_endpoint must be unchanged\n{yaml}" + ); + } + + #[test] + fn cold_format_intchunk_emits_format_and_derived_coldpart_endpoint() { + // When the deploy opts into the intchunk cold-part format, the + // emitted agent `cold:` block must carry `format: intchunk` and a + // `coldpart_endpoint:` derived from the fragment ship_endpoint + // (same merger host:port, `/ingest/coldpart` path). The + // ship_endpoint (fragment target) is still emitted unchanged. + let _env = crate::test_support::EnvVarGuard::set("ASAP_EDGE_FUSED", "1"); + let mut cfg = fused_asap_edge_cfg(); + cfg.cold_format = ColdFormat::Intchunk; + // cold_coldpart_endpoint left None ⇒ derive from ship_endpoint. + let yaml = emit_edge_yaml(&cfg, "ws://controller:4320/v1/opamp", "agent-1") + .expect("emit fused asap_edge ok"); + + let doc: serde_yaml::Value = + serde_yaml::from_str(&yaml).unwrap_or_else(|e| panic!("emitted YAML must parse: {e}\n{yaml}")); + let cold = doc + .get("processors") + .and_then(|p| p.get("asap_edge")) + .and_then(|p| p.get("cold")) + .expect("cold block present"); + assert_eq!( + cold.get("format").and_then(|v| v.as_str()), + Some("intchunk"), + "intchunk cold block must carry `format: intchunk`\n{yaml}" + ); + assert_eq!( + cold.get("coldpart_endpoint").and_then(|v| v.as_str()), + Some("http://gorilla-merger:10908/ingest/coldpart"), + "coldpart_endpoint must be derived from the ship_endpoint (\ + same merger host:port, /ingest/coldpart path)\n{yaml}" + ); + // The fragment ship_endpoint stays present (the agent still knows + // the fragment target; only the active format flips). + assert_eq!( + cold.get("ship_endpoint").and_then(|v| v.as_str()), + Some("http://gorilla-merger:10908/ingest/gorilla"), + "ship_endpoint must remain unchanged\n{yaml}" + ); + } + + #[test] + fn cold_format_intchunk_honours_explicit_coldpart_endpoint() { + // An explicit `cold_coldpart_endpoint` wins over the ship-endpoint + // derivation — lets a deploy point the cold-part tier at a + // different merger host if needed. + let _env = crate::test_support::EnvVarGuard::set("ASAP_EDGE_FUSED", "1"); + let mut cfg = fused_asap_edge_cfg(); + cfg.cold_format = ColdFormat::Intchunk; + cfg.cold_coldpart_endpoint = + Some("http://other-merger:10908/ingest/coldpart".into()); + let yaml = emit_edge_yaml(&cfg, "ws://controller:4320/v1/opamp", "agent-1") + .expect("emit fused asap_edge ok"); + + let doc: serde_yaml::Value = + serde_yaml::from_str(&yaml).unwrap_or_else(|e| panic!("emitted YAML must parse: {e}\n{yaml}")); + let cold = doc + .get("processors") + .and_then(|p| p.get("asap_edge")) + .and_then(|p| p.get("cold")) + .expect("cold block present"); + assert_eq!( + cold.get("coldpart_endpoint").and_then(|v| v.as_str()), + Some("http://other-merger:10908/ingest/coldpart"), + "explicit coldpart_endpoint must win over the derivation\n{yaml}" + ); + } + #[test] fn fused_asap_edge_tier_derives_from_archive_routing() { // Focused regression for the per-metric `tier` contract (companion @@ -5681,6 +5818,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, }; let yaml = emit_edge_yaml(&cfg, "ws://c/", "agent-1").expect("emit ok"); diff --git a/control_plane/src/emit/telegraf.rs b/control_plane/src/emit/telegraf.rs index 2b583af2..b4cd1d43 100644 --- a/control_plane/src/emit/telegraf.rs +++ b/control_plane/src/emit/telegraf.rs @@ -320,6 +320,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -339,6 +341,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -362,6 +366,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } @@ -517,6 +523,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: std::collections::HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, }; 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 cdca2fb7..7cdc1e89 100644 --- a/control_plane/src/emit/trait_def.rs +++ b/control_plane/src/emit/trait_def.rs @@ -213,6 +213,8 @@ mod tests { cold_ship_endpoint: None, cold_external_labels: Vec::new(), metric_to_sample_p: HashMap::new(), + cold_format: crate::physical::colored_dag::emitter::ColdFormat::default(), + cold_coldpart_endpoint: None, } } diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 75df3a1c..43bb2ae9 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -290,6 +290,59 @@ pub struct EdgeStageConfig { /// Empty map (default) ⇒ no metric carries sampling — backward-compat. #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub metric_to_sample_p: HashMap, + /// Cold-archive **wire format** the agent's `asapedgeprocessor` ships + /// its cold tier in. Two formats are merged in the agent: + /// + /// - [`ColdFormat::Fragment`] (default) — gorilla-XOR fragments, shipped + /// to [`cold_ship_endpoint`](Self::cold_ship_endpoint) (`/ingest/gorilla`). + /// - [`ColdFormat::Intchunk`] — the lossless intchunk cold-part format, + /// shipped to [`cold_coldpart_endpoint`](Self::cold_coldpart_endpoint) + /// (`/ingest/coldpart`). + /// + /// The L5 edge emitter writes a `cold.format` + `cold.coldpart_endpoint` + /// pair onto the agent `cold:` block ONLY when this is + /// [`ColdFormat::Intchunk`]. [`ColdFormat::Fragment`] (the default) + /// emits NEITHER key, so the agent's cold block stays byte-identical to + /// the pre-format emit (`ship_endpoint` only) — no behavior change when + /// unset. + #[serde(default, skip_serializing_if = "ColdFormat::is_default")] + pub cold_format: ColdFormat, + /// Cold-archive intchunk ingest URL — the gorilla-merger's coldpart + /// HTTP ingest endpoint (`http://gorilla-merger:10908/ingest/coldpart`). + /// Only emitted (and only meaningful) when + /// [`cold_format`](Self::cold_format) is [`ColdFormat::Intchunk`]. + /// + /// `None` ⇒ the emitter derives it from + /// [`cold_ship_endpoint`](Self::cold_ship_endpoint) by swapping the path + /// to `/ingest/coldpart` (same merger host:port as the fragment + /// endpoint), falling back to [`default_cold_coldpart_endpoint`] when + /// neither is set. Ignored entirely for [`ColdFormat::Fragment`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cold_coldpart_endpoint: Option, +} + +/// Cold-archive wire format the agent ships its cold tier in. See +/// [`EdgeStageConfig::cold_format`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ColdFormat { + /// gorilla-XOR fragments → `cold.ship_endpoint` (`/ingest/gorilla`). + /// The default — emits NO `format:`/`coldpart_endpoint:` keys, so the + /// agent cold block is byte-identical to the pre-format emit. + #[default] + Fragment, + /// Lossless intchunk cold-part → `cold.coldpart_endpoint` + /// (`/ingest/coldpart`). + Intchunk, +} + +impl ColdFormat { + /// `true` for the default ([`ColdFormat::Fragment`]). Drives the + /// `skip_serializing_if` on [`EdgeStageConfig::cold_format`] so an unset + /// format leaves the serialized config byte-identical to today. + pub fn is_default(&self) -> bool { + matches!(self, ColdFormat::Fragment) + } } /// Named default for [`EdgeStageConfig::cold_ship_endpoint`]. @@ -304,6 +357,30 @@ pub fn default_cold_ship_endpoint() -> String { "http://gorilla-merger:10908/ingest/gorilla".to_string() } +/// Named default for [`EdgeStageConfig::cold_coldpart_endpoint`]. +/// +/// The intchunk cold-part tier ships to the SAME gorilla-merger host:port +/// as the fragment tier, but on the `/ingest/coldpart` path (the fragment +/// path is `/ingest/gorilla`). Single source of truth so the emitter and +/// any construction site agree. Only consulted when +/// [`EdgeStageConfig::cold_format`] is [`ColdFormat::Intchunk`]. +pub fn default_cold_coldpart_endpoint() -> String { + "http://gorilla-merger:10908/ingest/coldpart".to_string() +} + +/// Derive a coldpart ingest URL from a fragment `ship_endpoint` by +/// swapping the trailing `/ingest/gorilla` path for `/ingest/coldpart` +/// (the merger host:port is shared between the two cold tiers). Falls back +/// to [`default_cold_coldpart_endpoint`] when the input doesn't carry the +/// expected fragment path, so a non-standard endpoint still yields a +/// well-formed coldpart target rather than a malformed one. +pub fn coldpart_endpoint_from_ship(ship_endpoint: &str) -> String { + match ship_endpoint.strip_suffix("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/ingest/gorilla") { + Some(host) => format!("{host}/ingest/coldpart"), + None => default_cold_coldpart_endpoint(), + } +} + /// Named default for [`EdgeStageConfig::cold_external_labels`]. /// /// One `cluster` label, read from `ASAP_CLUSTER` (default `asap-mvp`). @@ -596,6 +673,12 @@ impl Emitter for ThreeStageEmitter { cold_ship_endpoint: Some(default_cold_ship_endpoint()), cold_external_labels: default_cold_external_labels(), metric_to_sample_p: HashMap::new(), + // Cold-archive format defaults to gorilla-XOR fragments; the + // intchunk format (and its coldpart endpoint) is opted into by + // a deploy-info-bearing layer post-emit (same pattern as the + // cold endpoint above), keeping this layer deployment-agnostic. + cold_format: ColdFormat::default(), + cold_coldpart_endpoint: None, }; let mut backend_aggregations: Vec = Vec::new(); let mut gateway_processors: Vec = Vec::new();