From 074198d1ea4f35a057d3f136f68dfad7ae716085 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 24 May 2026 19:38:16 -0600 Subject: [PATCH] feat(emit): per-metric tier in fused asap_edge config (#46 follow-up) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each `metrics[]` entry in the fused `asap_edge` agent config now carries a `tier` ∈ {warm, both, cold}, derived from the plan routing already on `EdgeStageConfig` (no hardcoded metric→tier map): * warm signal — the metric produces a warm entry (sketch family in `metric_to_family` or a Sum-by aggregate). * cold signal — the metric is in `archive_tier_metrics` (the plan's exact/archive routing decision the legacy gorillas3 path consumed). tier = both (warm+cold), warm (warm only), cold (cold only), defaulting to `both` when no signal is determinable (safe — preserves archival, matching the processor's unset-tier default). This lets the fused agent's cold tier archive ONLY what the plan says needs archiving (companion to the ASAPCollector asapedgeprocessor `tier` field). For the MVP workload: `http_requests_total` (warm sum/rate + exact `count(...)` archive query) → both; the sketch-only metrics (`http_requests_total_latency_ms`, `request_size_bytes`, `unique_users_per_min`, `top_endpoint_qps`, `endpoint_request_freq`) → warm. Tests: extend the fused emit test to assert tier=both/warm per metric, plus a focused tier-derivation test; both route `ASAP_EDGE_FUSED` through the shared env-lock harness (#318). Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/emit/stage_config.rs | 159 +++++++++++++++++++++++++ 1 file changed, 159 insertions(+) diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index a95951ed5..119fc4228 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -1681,6 +1681,55 @@ fn emit_edge_yaml_asap_edge( let mut metric_entries: Vec = Vec::new(); + // ── Per-metric storage tier (issue #46 follow-up; companion to the + // ASAPCollector asapedgeprocessor `tier` field) ───────────────────────── + // + // Each emitted `metrics[]` entry carries a `tier` ∈ {warm, both, cold} + // telling the fused agent which storage tiers to feed the metric into: + // + // * `warm` — warm sketch/aggregation ONLY; the metric is NOT + // cold-archived by the agent's Gorilla encoder. + // * `both` — warm sketch/agg AND cold gorilla archive. + // * `cold` — cold gorilla archive ONLY; no warm sketch/agg. + // + // We DERIVE the tier from the SAME plan routing the rest of this emit + // reads — no hardcoded metric→tier table — so it generalises to any + // workload: + // + // * "warm" signal — the metric produces a warm entry below (a + // Sum-by aggregate from `cumulative_counter_metrics` + + // `metric_to_grouping_labels`, or a sketch family from + // `metric_to_family`). This is exactly the routing that lands a + // metric in the warm sketch/agg tier. + // * "cold" signal — the metric is in `archive_tier_metrics`, the + // plan's archive-routing decision (an exact / archive query forces + // the Gorilla object-store archive; the routing that the legacy + // `gorillas3` processor consumed). + // + // tier = both when a metric has BOTH signals (e.g. `http_requests_total` + // — warm `sum by (zone)` AND an exact `count(...)` archive query), warm + // when only the warm signal is present (the sketch-only quantile / HLL / + // topk / rate metrics), cold when only the archive signal is present. + // When neither is determinable the metric defaults to `both` (safe — + // preserves archival), matching the processor's unset-tier default. + let cold_set: std::collections::BTreeSet<&str> = cfg + .archive_tier_metrics + .iter() + .map(|a| a.metric.as_str()) + .collect(); + let tier_for = |metric: &str, warm: bool| -> &'static str { + let cold = cold_set.contains(metric); + match (warm, cold) { + (true, true) => "both", + (true, false) => "warm", + (false, true) => "cold", + // No routing signal at all — default to `both` so the agent + // keeps archiving (preserves data); the processor treats an + // unset tier the same way. + (false, false) => "both", + } + }; + // Sum-family entries — same predicate as ASAPCollector#403's // edge-aggregate selection: Sum-role metric WITH grouping labels and // NOT mapped to a sketch family. @@ -1705,6 +1754,11 @@ fn emit_edge_yaml_asap_edge( e.insert("family".into(), Value::String("sum".to_string())); let by: Vec = labels.into_iter().map(Value::String).collect(); e.insert("aggregate_by".into(), Value::Sequence(by)); + // Sum aggregate IS a warm entry → warm signal = true. + e.insert( + "tier".into(), + Value::String(tier_for(metric, true).to_string()), + ); metric_entries.push(Value::Mapping(e)); } @@ -1776,6 +1830,11 @@ fn emit_edge_yaml_asap_edge( } } } + // Sketch family IS a warm entry → warm signal = true. + e.insert( + "tier".into(), + Value::String(tier_for(metric, true).to_string()), + ); metric_entries.push(Value::Mapping(e)); } } @@ -5351,6 +5410,15 @@ mod tests { .map(|v| v.as_str().unwrap().to_string()) .collect(); assert_eq!(by, vec!["zone".to_string()], "sum aggregate_by\n{yaml}"); + // tier=both: `http_requests_total` is in `archive_tier_metrics` + // (the exact `count(...)` archive query) AND has a warm sum-by + // aggregate, so the agent must feed BOTH the warm sum and the + // cold gorilla archive. + assert_eq!( + sum_e.get("tier").and_then(|v| v.as_str()), + Some("both"), + "archive + warm metric must emit tier=both\n{yaml}" + ); // Sketch entries + params. let dd = entry_for("http_requests_total_latency_ms"); @@ -5376,6 +5444,24 @@ 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)); + // tier=warm: the five sketch-only metrics are NOT in + // `archive_tier_metrics` (no exact/archive query), so the agent + // builds their warm sketch ONLY and does NOT cold-archive them — + // exactly the bandwidth win this contract buys. + for sketch_only in [ + "http_requests_total_latency_ms", + "request_size_bytes", + "unique_users_per_min", + "top_endpoint_qps", + "endpoint_request_freq", + ] { + assert_eq!( + entry_for(sketch_only).get("tier").and_then(|v| v.as_str()), + Some("warm"), + "sketch-only metric {sketch_only} must emit tier=warm\n{yaml}" + ); + } + // 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 @@ -5439,6 +5525,79 @@ mod tests { assert!(yaml.contains("otlp/backend:"), "{yaml}"); } + #[test] + fn fused_asap_edge_tier_derives_from_archive_routing() { + // Focused regression for the per-metric `tier` contract (companion + // to the ASAPCollector asapedgeprocessor `tier` field). The tier is + // DERIVED from the plan routing already on `EdgeStageConfig`: + // * warm signal — the metric has a warm entry (sketch family in + // `metric_to_family` or Sum-by aggregate). + // * cold signal — the metric is in `archive_tier_metrics` (the + // plan's exact/archive routing decision). + // tier = both (warm+cold), warm (warm only), defaulting to both + // when no signal is present. + // + // `ASAP_EDGE_FUSED` is process-global; set it under the crate-wide + // env lock (the #318 shared harness) so a parallel thread can't + // observe this test's setenv as its own input. + let _env = crate::test_support::EnvVarGuard::set("ASAP_EDGE_FUSED", "1"); + + // Two metrics: a sketch-only one (warm) and one that is BOTH + // sketched AND archived (both). The archive set is the precise + // plan signal — only `archived_metric` is in it. + let mut metric_to_family: HashMap> = + HashMap::new(); + metric_to_family.insert("sketch_only_metric".into(), [SketchKind::Kll].into()); + metric_to_family.insert("archived_metric".into(), [SketchKind::DDSketch].into()); + + let cfg = EdgeStageConfig { + source_metric: None, + label_filters: Vec::new(), + window_secs: Some(60), + sketch_processors: Vec::new(), + exporter_target: ExportTarget::Endpoint("backend:4317".into()), + prometheus_archive_metrics: Vec::new(), + archive_tier_metrics: vec![ArchiveTierMetric { + metric: "archived_metric".into(), + window_secs: Some(60), + }], + warm_passthrough_metrics: Vec::new(), + metric_to_family, + 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/", "agent-1").expect("emit ok"); + let doc: serde_yaml::Value = serde_yaml::from_str(&yaml).expect("parse"); + let metrics = doc + .get("processors") + .and_then(|p| p.get("asap_edge")) + .and_then(|a| a.get("metrics")) + .and_then(|v| v.as_sequence()) + .expect("asap_edge.metrics seq"); + let tier_of = |name: &str| -> Option { + metrics + .iter() + .find(|e| e.get("metric").and_then(|m| m.as_str()) == Some(name)) + .and_then(|e| e.get("tier")) + .and_then(|v| v.as_str()) + .map(str::to_string) + }; + + assert_eq!( + tier_of("sketch_only_metric").as_deref(), + Some("warm"), + "sketch-only (warm signal, no archive routing) must be tier=warm\n{yaml}" + ); + assert_eq!( + tier_of("archived_metric").as_deref(), + Some("both"), + "sketched + archived metric must be tier=both\n{yaml}" + ); + } + #[test] fn fused_gate_off_keeps_routing_shape() { // Without the env gate the canonical routing-connector shape is