From b88baf462355f64efb187578550ef99946221996 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 18 May 2026 18:05:29 -0600 Subject: [PATCH] fix(emit): inject cumulativetodelta upstream of agent routing for Counter metrics (closes #298) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fake-exporter's `Float64Counter` instruments default to OTel-SDK **cumulative** temporality — each export carries the running lifetime value, not the per-export delta. The asap-otel agent's 5-sketch routing pipeline previously had no `cumulativetodelta` processor, so the backend's `SumAccumulator::update` (sum-of-deltas) was fed cumulatives and computed `Σ-of-cumulatives-in-window` — a quadratic-in-time blowup. The replay path's outer sum across the lookback then made it cubic, so `sum by (zone) (http_requests_total)` on the asap tier returned ~300× the b0 baseline. Fix: - New `EdgeStageConfig::cumulative_counter_metrics: Vec` field, populated from a new `collect_cumulative_counter_metrics()` helper that walks the WorkloadStore picking up every metric whose workload query classifies as `AggRole::Sum` (bare-selector / sum / rate / increase / sum_over_time / irate). - `emit_edge_yaml_5sketch_routing` declares a `cumulativetodelta` processor with `include.metrics = [...]` (`match_type: strict`, sort-stable for byte-stable YAML) when the list is non-empty, and inserts it as the FIRST processor in the entry `metrics:` pipeline so every routed copy of each listed metric reaches the routing connector with delta temporality. - Bootstrap (`main::emit_bootstrap_typed`) and OpAMP replan (`replan::Replanner::try_emit_typed_edge_yaml_for_workload`) both wire the new field via the new helper, mirroring the existing `metric_to_family` / `metric_to_grouping_labels` stitches. - Strict matching keeps the processor a no-op for any unrelated metric (quantile gauges like `http_requests_total_latency_ms` pass through unchanged). 6 new tests pin: * presence/absence of the processor by list non-emptiness * processor runs FIRST on the entry pipeline (before routing connector fan-out) * include-metrics list is sorted (byte-stable YAML across runs) * helper classifies Sum-role entries correctly, dedupes, and skips quantile/cardinality entries Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/emit/mod.rs | 136 ++++++++++++ control_plane/src/emit/otap.rs | 3 + control_plane/src/emit/stage_config.rs | 209 +++++++++++++++++- control_plane/src/emit/telegraf.rs | 4 + control_plane/src/emit/trait_def.rs | 1 + control_plane/src/main.rs | 10 + .../src/physical/colored_dag/emitter.rs | 28 +++ control_plane/src/replan.rs | 8 + 8 files changed, 396 insertions(+), 3 deletions(-) diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index 3946eca8..2792e341 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -356,6 +356,62 @@ pub fn collect_metric_to_grouping_labels( out } +/// Issue #298 — sibling of [`collect_metric_to_family`] / +/// [`collect_metric_to_grouping_labels`]: walk every registry entry and +/// return the deduped list of metrics whose workload(s) classify as +/// [`crate::workload::AggRole::Sum`] — bare-selector / `sum` / `rate` +/// / `increase` / `sum_over_time` / `irate`. These are the +/// Counter-shaped metrics whose OTel SDK emission defaults to +/// **cumulative** temporality and must be converted to **delta** before +/// the backend's `SumAccumulator` folds them, otherwise the +/// per-window sum is `Σ-of-cumulatives-in-window` (quadratic-in-time +/// blowup; cubic for instant `sum by (zone) (counter)` reads). +/// +/// Drops directly into `EdgeStageConfig::cumulative_counter_metrics`, +/// which the 5-sketch routing emitter consumes to declare a +/// `cumulativetodelta` processor with `include.metrics = [...]` on the +/// entry pipeline. Empty list ⇒ no processor emitted (backward-compat +/// for quantile-only / sketch-only plans). +/// +/// **A metric is included iff ANY of its registered roles classifies +/// as Sum**. This is the conservative direction: a metric with even +/// one Sum-shaped query needs delta conversion for that query to be +/// correct, and the OTel processor's `match_type: strict` filter then +/// gates which metrics the processor actually rewrites (every other +/// metric on the wire is a no-op pass-through). Gauge data points +/// carry no aggregation_temporality at all (it's a Counter-only +/// concept), so the processor leaves them untouched if a metric is +/// also used as a gauge elsewhere. +pub fn collect_cumulative_counter_metrics( + registry: &WorkloadRegistry, + workload_store: &WorkloadStore, +) -> Vec { + use crate::workload::{derive_agg_role, AggRole}; + let mut seen: std::collections::BTreeSet = std::collections::BTreeSet::new(); + for entry in registry.entries() { + // `derive_agg_role` reads the WorkloadEntry directly (query + // string + family override), not the lowered QueryWorkload, so + // we classify the registry entry. We still consult the + // workload_store to confirm the metric was successfully + // pre-populated (matching the contract of the sibling + // collectors) — silent skips for entries that failed the + // pre-pop keep the emit aligned with what the backend actually + // knows about. + if workload_store + .get_all_for_metric(&entry.metric_name) + .into_iter() + .next() + .is_none() + { + continue; + } + if derive_agg_role(entry) == AggRole::Sum { + seen.insert(entry.metric_name.clone()); + } + } + seen.into_iter().collect() +} + #[cfg(test)] mod runtime_tests { use super::*; @@ -414,6 +470,7 @@ mod runtime_tests { 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(), }; let collector = emit_for_runtime( @@ -448,6 +505,7 @@ mod runtime_tests { 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(), }; let yaml = emit_for_runtime( AgentRuntime::AsapOtap, @@ -480,6 +538,7 @@ mod runtime_tests { 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(), }; let toml = emit_for_runtime( AgentRuntime::AsapTelegraf, @@ -719,6 +778,7 @@ mod runtime_tests { SketchKind::DDSketch, )]), metric_to_grouping_labels: std::collections::HashMap::new(), + cumulative_counter_metrics: Vec::new(), }; edge_cfg.metric_to_grouping_labels = collect_metric_to_grouping_labels(®istry, &store); @@ -740,4 +800,80 @@ mod runtime_tests { "empty keep_keys would strip all attrs and break per-zone sid splitting\n{yaml_out}", ); } + + // ── Issue #298 — collect_cumulative_counter_metrics ──────────────────── + + /// Workload with mixed roles — a bare counter selector, a `sum by + /// (...)` over a counter, and a quantile gauge. Only the first two + /// classify as `AggRole::Sum`; the gauge query is `AggRole::Quantile` + /// and must NOT appear in the output. The two Sum entries refer to + /// the SAME metric (`http_requests_total`), so the helper dedupes. + #[test] + fn issue298_collect_cumulative_counter_metrics_picks_sum_role_dedup() { + let yaml = r#" +- metric_name: http_requests_total + query_string: "http_requests_total" + accuracy_sla: 0.0 + assign_to_role: agent +- metric_name: http_requests_total + query_string: "sum by (zone) (http_requests_total)" + accuracy_sla: 0.0 + assign_to_role: gateway +- metric_name: http_requests_total_latency_ms + query_string: "quantile_over_time(0.99, http_requests_total_latency_ms[30s])" + accuracy_sla: 0.01 + assign_to_role: agent +- metric_name: endpoint_request_freq + query_string: "rate(endpoint_request_freq[5m])" + accuracy_sla: 0.05 + assign_to_role: agent +"#; + let entries: Vec = + serde_yaml::from_str(yaml).expect("parse workload yaml"); + let registry = crate::workload::WorkloadRegistry::from_entries(entries); + let store = WorkloadStore::new(); + populate_store_from_registry(®istry, &store); + + let counters = collect_cumulative_counter_metrics(®istry, &store); + assert_eq!( + counters, + vec![ + "endpoint_request_freq".to_string(), + "http_requests_total".to_string(), + ], + "expected the Sum-role metrics deduped + sorted; the \ + quantile_over_time entry on http_requests_total_latency_ms \ + must NOT appear (it's AggRole::Quantile)" + ); + } + + /// Workload with zero Sum-shaped entries (all quantile / cardinality) + /// produces an empty list — the emitter then skips the + /// `cumulativetodelta` processor entirely (backward-compat for + /// quantile-only deployments). + #[test] + fn issue298_collect_cumulative_counter_metrics_empty_for_quantile_only_workload() { + let yaml = r#" +- metric_name: http_requests_total_latency_ms + query_string: "quantile_over_time(0.99, http_requests_total_latency_ms[30s])" + accuracy_sla: 0.01 + assign_to_role: agent +- metric_name: unique_users_per_min + query_string: "count(unique_users_per_min)" + accuracy_sla: 0.02 + assign_to_role: agent +"#; + let entries: Vec = + serde_yaml::from_str(yaml).expect("parse workload yaml"); + let registry = crate::workload::WorkloadRegistry::from_entries(entries); + let store = WorkloadStore::new(); + populate_store_from_registry(®istry, &store); + + let counters = collect_cumulative_counter_metrics(®istry, &store); + assert!( + counters.is_empty(), + "quantile / cardinality entries must not be classified as \ + cumulative counters; got {counters:?}" + ); + } } diff --git a/control_plane/src/emit/otap.rs b/control_plane/src/emit/otap.rs index 755913fd..5d62da02 100644 --- a/control_plane/src/emit/otap.rs +++ b/control_plane/src/emit/otap.rs @@ -395,6 +395,7 @@ mod tests { 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(), } } @@ -410,6 +411,7 @@ mod tests { 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(), } } @@ -429,6 +431,7 @@ mod tests { 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(), } } diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 2c816e16..c39b1f3a 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -1061,6 +1061,69 @@ fn emit_edge_yaml_5sketch_routing( .context("parse memory_limiter processor block")?; processors.insert("memory_limiter".to_string(), memory_limiter_block); + // ── cumulativetodelta processor — Issue #298 ─────────────────────────── + // + // The OTel SDK's `Counter` instruments default to **cumulative** + // temporality: every export carries the running lifetime value of + // the counter, not the per-export delta. Backend's `SumAccumulator` + // (`data_plane/src/precompute_engine/operators/sum_accumulator.rs`) + // naïvely sums every incoming value into the per-window state — fed + // cumulative data, it computes `Σ-of-cumulatives-in-window`, a + // quadratic-in-time blowup. The replay path then re-sums those + // inflated per-window values across the lookback range → cubic + // blowup for instant `sum by (zone) (counter)` queries. + // (Observed: ~300× the baseline pre-fix; see Issue #298.) + // + // Fix: register the contrib build's `cumulativetodelta` processor + // with an `include.metrics` allowlist of the workload's + // Counter-shaped metrics (sourced from + // `collect_cumulative_counter_metrics` — every workload entry whose + // query classifies as `AggRole::Sum`), and run it as the FIRST + // processor in the entry (`metrics:`) pipeline so EVERY routed copy + // of each listed metric reaches the routing connector with delta + // temporality. + // + // `match_type: strict` keeps the processor a no-op for any other + // metric (gauges like `http_requests_total_latency_ms` pass through + // unchanged — quantile / histogram workloads keep their wire shape). + // + // Why entry pipeline (not per-family pipeline): the routing + // connector dispatches on `metric.name`; running the conversion + // upstream of the connector means every per-family pipeline AND the + // `raw_passthrough` default both see deltas. Per-pipeline placement + // would duplicate work and risk double-conversion on pipelines that + // a future plan fans the metric into. + // + // Empty `cumulative_counter_metrics` ⇒ no processor declared, no + // entry-pipeline processor list — backward-compat for + // quantile-only / sketch-only plans that never declare a counter. + let needs_cumulativetodelta = !cfg.cumulative_counter_metrics.is_empty(); + if needs_cumulativetodelta { + // Deterministic order so the emitted YAML is stable across + // controller runs — mirrors the BTreeMap-not-HashMap rationale + // on `CollectorYaml`. The agent's opampextension byte-compares + // pushed configs; an unsorted include list would force an + // apply+restart on every push of the same semantic plan. + let mut sorted_metrics: Vec<&String> = + cfg.cumulative_counter_metrics.iter().collect(); + sorted_metrics.sort(); + // YAML indentation note: `metrics` and `match_type` are both + // direct children of `include` (not of each other). The + // `include.metrics` list entries indent two more spaces under + // `metrics:`. Get this wrong and serde_yaml rejects the block + // with "did not find expected key" at parse time. + let mut metrics_yaml = String::new(); + for m in &sorted_metrics { + metrics_yaml.push_str(&format!(" - \"{m}\"\n")); + } + let cumulativetodelta_yaml = format!( + "include:\n metrics:\n{metrics_yaml} match_type: strict\n" + ); + let cumulativetodelta_block: Value = serde_yaml::from_str(&cumulativetodelta_yaml) + .context("parse cumulativetodelta processor block")?; + processors.insert("cumulativetodelta".to_string(), cumulativetodelta_block); + } + // ── Exporters ────────────────────────────────────────────────────────── // Edge → asapquery-backend OTLP ingest (see emit_edge_yaml for the // gateway-less rationale). @@ -1196,13 +1259,23 @@ fn emit_edge_yaml_5sketch_routing( // Entry pipeline — receivers: [otlp], exporters: [routing] // (`routing` here is the connector, used as exporter for the entry - // stage). NO processors on the entry pipeline; the connector is - // responsible for fan-out. + // stage). The processor list is normally empty (the connector owns + // fan-out), but Issue #298 requires `cumulativetodelta` to run + // BEFORE the connector so EVERY routed copy of a Counter-shaped + // metric reaches the downstream pipelines with delta temporality. + // Putting the conversion here (not per-family) avoids duplicating + // the conversion across the per-family pipelines AND the + // `raw_passthrough` default, and stops fan-in-from-multiple-routes + // double-conversion. + let mut entry_processors: Vec = Vec::new(); + if needs_cumulativetodelta { + entry_processors.push("cumulativetodelta".to_string()); + } pipelines.insert( "metrics".to_string(), Pipeline { receivers: vec!["otlp".into()], - processors: Vec::new(), + processors: entry_processors, exporters: vec!["routing".to_string()], }, ); @@ -1799,6 +1872,7 @@ mod tests { warm_passthrough_metrics: Vec::new(), metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), + cumulative_counter_metrics: Vec::new(), } } @@ -2770,6 +2844,7 @@ mod tests { warm_passthrough_metrics: Vec::new(), metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), + cumulative_counter_metrics: Vec::new(), }; let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); @@ -3094,6 +3169,7 @@ mod tests { warm_passthrough_metrics: Vec::new(), metric_to_family, metric_to_grouping_labels: HashMap::new(), + cumulative_counter_metrics: Vec::new(), } } @@ -3993,4 +4069,131 @@ mod tests { ); } + // ── Issue #298: cumulativetodelta on counter metrics ────────────────── + // + // OTel SDK `Counter` instruments default to cumulative temporality. + // Backend's `SumAccumulator::update` is sum-of-deltas — fed + // cumulative data it returns `Σ-of-cumulatives-in-window` (quadratic + // in time; cubic after the reducer's outer sum across the lookback + // range). The fix is to inject `cumulativetodelta` upstream of the + // routing connector, scoped to the workload's Counter-shaped + // metrics. Tests below pin: + // * presence of the processor declaration when the list is + // non-empty, with the listed metrics as the `include` filter, + // and the entry pipeline running it FIRST; + // * absence (legacy quantile-only behaviour) when the list is + // empty — backward-compat; + // * sort-stability so the emitted YAML is byte-stable across + // planner runs (HashMap iteration drift would otherwise trip + // the agent's no-op apply check). + + #[test] + fn issue298_cumulativetodelta_emitted_when_counter_metrics_present() { + let mut cfg = five_sketch_edge_cfg(); + cfg.cumulative_counter_metrics = vec![ + "http_requests_total".to_string(), + "endpoint_request_freq".to_string(), + ]; + let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); + assert!( + yaml.contains("cumulativetodelta:"), + "expected cumulativetodelta processor declaration when \ + cumulative_counter_metrics is non-empty\n{yaml}" + ); + // Strict match_type so the processor stays a no-op for metrics + // not in the include list (gauges, quantile metrics). + assert!( + yaml.contains("match_type: strict"), + "cumulativetodelta processor must use strict include matching\n{yaml}" + ); + // Both metrics appear under the include.metrics list. The YAML + // serializer drops the redundant quotes on simple identifiers + // (`- endpoint_request_freq`); we assert on the bare list-item + // form, which is what the agent's confmap parser will accept. + for m in ["http_requests_total", "endpoint_request_freq"] { + assert!( + yaml.contains(&format!("- {m}\n")) || yaml.contains(&format!("- \"{m}\"\n")), + "expected metric {m} as a list item in include.metrics\n{yaml}" + ); + } + } + + #[test] + fn issue298_cumulativetodelta_runs_first_on_entry_pipeline() { + let mut cfg = five_sketch_edge_cfg(); + cfg.cumulative_counter_metrics = vec!["http_requests_total".to_string()]; + let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); + + // Locate the entry `metrics:` pipeline (NOT `metrics/...`) under + // service.pipelines and verify its `processors:` list contains + // `cumulativetodelta` ahead of any other processor (it's the + // only entry-pipeline processor, so checking presence on the + // entry block is sufficient + the section's `exporters: [routing]` + // anchor proves we matched the entry pipeline). + let pipelines_idx = yaml.find("pipelines:").expect("pipelines:"); + let after = &yaml[pipelines_idx..]; + let entry_marker = "\n metrics:\n"; + let entry_idx = after.find(entry_marker).expect("entry pipeline"); + let entry_after = &after[entry_idx + entry_marker.len()..]; + let next_metric_pipeline = entry_after + .find("\n metrics/") + .map(|x| x) + .unwrap_or(entry_after.len()); + let section = &entry_after[..next_metric_pipeline]; + assert!( + section.contains("- cumulativetodelta"), + "entry pipeline must list cumulativetodelta as a processor\n{section}" + ); + assert!( + section.contains("- routing"), + "entry pipeline must keep exporters: [routing]\n{section}" + ); + } + + #[test] + fn issue298_cumulativetodelta_omitted_when_no_counter_metrics() { + // five_sketch_edge_cfg() leaves cumulative_counter_metrics + // empty by default — verify the processor is NOT declared and + // the entry pipeline's processors list stays empty (backward- + // compat for quantile-only deployments). + let cfg = five_sketch_edge_cfg(); + assert!( + cfg.cumulative_counter_metrics.is_empty(), + "test precondition: default cfg has no counter metrics" + ); + let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); + assert!( + !yaml.contains("cumulativetodelta"), + "cumulativetodelta processor must NOT be emitted when \ + cumulative_counter_metrics is empty\n{yaml}" + ); + } + + #[test] + fn issue298_cumulativetodelta_include_list_is_sorted() { + // HashMap iteration is not order-stable — but the agent's + // opampextension byte-level no-op check would otherwise apply + // + restart on every push of the same semantic config. Mirrors + // the BTreeMap-not-HashMap rationale on `CollectorYaml`. + let mut cfg = five_sketch_edge_cfg(); + cfg.cumulative_counter_metrics = vec![ + "zzz_counter".to_string(), + "aaa_counter".to_string(), + "mmm_counter".to_string(), + ]; + let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); + // Quotes get stripped by serde_yaml for simple identifiers; + // probe both forms so the assertion survives either output. + let find_any = |needle_a: &str, needle_b: &str| { + yaml.find(needle_a).or_else(|| yaml.find(needle_b)) + }; + let a_idx = find_any("- aaa_counter\n", "- \"aaa_counter\"\n").expect("aaa_counter"); + let m_idx = find_any("- mmm_counter\n", "- \"mmm_counter\"\n").expect("mmm_counter"); + let z_idx = find_any("- zzz_counter\n", "- \"zzz_counter\"\n").expect("zzz_counter"); + assert!( + a_idx < m_idx && m_idx < z_idx, + "include.metrics list must be sorted for byte-stable YAML \ + (a={a_idx} m={m_idx} z={z_idx})\n{yaml}" + ); + } } diff --git a/control_plane/src/emit/telegraf.rs b/control_plane/src/emit/telegraf.rs index f3a4f48c..b3ff11bc 100644 --- a/control_plane/src/emit/telegraf.rs +++ b/control_plane/src/emit/telegraf.rs @@ -316,6 +316,7 @@ mod tests { 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(), } } @@ -331,6 +332,7 @@ mod tests { 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(), } } @@ -350,6 +352,7 @@ mod tests { 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(), } } @@ -501,6 +504,7 @@ mod tests { 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(), }; 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 2302ae10..b50bdd85 100644 --- a/control_plane/src/emit/trait_def.rs +++ b/control_plane/src/emit/trait_def.rs @@ -209,6 +209,7 @@ mod tests { warm_passthrough_metrics: Vec::new(), metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), + cumulative_counter_metrics: Vec::new(), } } diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index a80fa473..0dbb3a8f 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -1270,6 +1270,16 @@ async fn emit_bootstrap_typed( &st.workload_registry, &st.workload_store, ); + // Issue #298 — companion stitch: list of Counter-shaped metrics + // the agent must run through `cumulativetodelta` upstream of the + // routing connector. Without this, the OTel SDK's default + // cumulative-temporality Counter export inflates the backend's + // per-window SumAccumulator into Σ-of-cumulatives, breaking + // `sum by (zone) (http_requests_total)` (~300× baseline pre-fix). + edge_cfg.cumulative_counter_metrics = emit::collect_cumulative_counter_metrics( + &st.workload_registry, + &st.workload_store, + ); // Issue #2: thread X-Agent-ID into the opamp block. Bootstrap GET // is per-agent when `pinned_agent_id` is set (the agent's own diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 33f1d541..85b20ff2 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -198,6 +198,33 @@ pub struct EdgeStageConfig { /// `grouping_labels` contract. #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub metric_to_grouping_labels: HashMap>, + /// Issue #298 — metrics whose OTel datapoints arrive with + /// **cumulative** temporality (OTel SDK's default for `Counter` + /// instruments) and need to be converted to **delta** before the + /// backend's `SumAccumulator` folds them into per-window sums. + /// + /// When non-empty, the 5-sketch routing emitter declares a + /// `cumulativetodelta` processor with `include.metrics = [...]` and + /// inserts it as the FIRST processor in the entry (`metrics:`) + /// pipeline so every routed copy of each listed metric goes through + /// the conversion. The processor matches on `metric.name` + /// (strict), so unrelated metrics flow through unchanged — quantile + /// gauges (`http_requests_total_latency_ms`) keep their wire shape. + /// + /// Sourced from [`crate::emit::collect_cumulative_counter_metrics`]: + /// any metric whose workload entry classifies as + /// [`crate::workload::AggRole::Sum`] (bare-selector / `sum` / + /// `rate` / `increase` / `sum_over_time` / `irate`). Without the + /// conversion, the data plane's `SumAccumulator` re-sums each + /// cumulative carry-value within and across windows, producing a + /// quadratic-in-time blowup (observed: `sum by (zone) + /// (http_requests_total)` returned ~300× baseline pre-fix). + /// + /// Empty list (default) ⇒ no `cumulativetodelta` processor is + /// emitted; backward-compat for plans that never declare a counter + /// metric (e.g. quantile-only workloads). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub cumulative_counter_metrics: Vec, } /// Phase 3.2.5 — one archive-tier metric the agent should land in @@ -472,6 +499,7 @@ impl Emitter for ThreeStageEmitter { warm_passthrough_metrics: Vec::new(), metric_to_family: HashMap::new(), metric_to_grouping_labels: HashMap::new(), + cumulative_counter_metrics: Vec::new(), }; let mut backend_aggregations: Vec = Vec::new(); let mut gateway_processors: Vec = Vec::new(); diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index b1b3efbd..d198ee78 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -281,6 +281,14 @@ impl Replanner { // OTTL processor in front of every sketch pipeline. edge_cfg.metric_to_grouping_labels = crate::emit::collect_metric_to_grouping_labels(registry, &self.workload_store); + // Issue #298 — companion stitch: Counter-shaped metrics + // that need `cumulativetodelta` upstream of the routing + // connector. Mirrors the bootstrap stitch in + // `main::emit_bootstrap_typed` so OpAMP-pushed re-plans + // carry the same processor declaration the first-connect + // bootstrap YAML did. + edge_cfg.cumulative_counter_metrics = + crate::emit::collect_cumulative_counter_metrics(registry, &self.workload_store); } // OpAMP `on_connect` doesn't expose the agent's runtime