diff --git a/control_plane/src/emit/agent.rs b/control_plane/src/emit/agent.rs index 7cce9cc7..ef64018a 100644 --- a/control_plane/src/emit/agent.rs +++ b/control_plane/src/emit/agent.rs @@ -137,7 +137,15 @@ fn build_processor_block(cfg: &AgentCollectorConfig) -> Value { if cfg.mode == ProcessorMode::Window { if let Some(wd) = cfg.window_duration { - m.insert("window_duration".into(), Value::String(format_duration(wd))); + // MVP blocker B4: clamp `window_duration` to [5, 60] so + // the legacy agent emitter matches the typed L5 emitter's + // bounds — without this, a `[5m]` workload landing here + // mints a 300s sketch window whose closed answer never + // falls inside the user's replay range. + let clamped = super::stage_config::clamp_window_secs(Some(wd.as_secs())) + .map(std::time::Duration::from_secs) + .unwrap_or(wd); + m.insert("window_duration".into(), Value::String(format_duration(clamped))); } } @@ -303,9 +311,50 @@ mod tests { #[test] fn contains_window_duration() { let yaml = generate_agent_collector_config(&ddsketch_cfg(), "ws://ctrl:4320/v1/opamp").unwrap(); + // MVP blocker B4: the fixture's 5m window clamps to 60s + // (`MAX_WINDOW_SECS`). Assert on the clamped form — a window + // larger than 60s would put the sketch close outside any + // sensible replay range. Pre-B4 this test asserted "5m". assert!( - yaml.contains("5m"), - "YAML should contain window_duration\n{yaml}" + yaml.contains("window_duration: 1m") || yaml.contains("window_duration: 60s"), + "YAML should contain clamped window_duration (1m / 60s)\n{yaml}" + ); + } + + #[test] + fn clamps_oversize_window_to_max() { + let mut cfg = ddsketch_cfg(); + cfg.window_duration = Some(Duration::from_secs(3600)); // 1h + let yaml = generate_agent_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap(); + assert!( + !yaml.contains("window_duration: 1h"), + "1h window must clamp to MAX_WINDOW_SECS, not pass through\n{yaml}" + ); + assert!( + yaml.contains("window_duration: 1m") || yaml.contains("window_duration: 60s"), + "clamped window must be 60s\n{yaml}" + ); + } + + #[test] + fn clamps_undersize_window_to_min() { + let mut cfg = ddsketch_cfg(); + cfg.window_duration = Some(Duration::from_secs(1)); // 1s + let yaml = generate_agent_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap(); + assert!( + yaml.contains("window_duration: 5s"), + "1s window must clamp UP to MIN_WINDOW_SECS=5s\n{yaml}" + ); + } + + #[test] + fn preserves_window_inside_clamp_range() { + let mut cfg = ddsketch_cfg(); + cfg.window_duration = Some(Duration::from_secs(30)); + let yaml = generate_agent_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap(); + assert!( + yaml.contains("window_duration: 30s"), + "30s window is inside [5, 60] and must pass through verbatim\n{yaml}" ); } diff --git a/control_plane/src/emit/asapquery_backend.rs b/control_plane/src/emit/asapquery_backend.rs index 27c63930..f072a5ad 100644 --- a/control_plane/src/emit/asapquery_backend.rs +++ b/control_plane/src/emit/asapquery_backend.rs @@ -41,9 +41,18 @@ use crate::types::{AgentCollectorConfig, CollectionPlan, SketchType}; /// serialization fails. pub fn generate_streaming_config_yaml(metric: &str, plan: &CollectionPlan) -> Result { let agg = &plan.agent_config; + // MVP blocker B4: clamp the workload's `window_duration` to + // `[MIN_WINDOW_SECS, MAX_WINDOW_SECS]` so the legacy YAML emit + // matches the typed L5 JSON emit's `windowSize` clamp. Without + // this, the legacy and typed paths can disagree (e.g. typed + // clamps `[5m]` → 60, legacy passes 300 → backend reducer keys + // a 300s window the agent never closes). let window_secs = agg .window_duration - .map(|d: Duration| d.as_secs()) + .map(|d: Duration| { + super::stage_config::clamp_window_secs(Some(d.as_secs())) + .expect("clamp preserves Some") + }) .unwrap_or(0); if window_secs == 0 { anyhow::bail!( diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index 50088d73..602506a7 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -497,13 +497,20 @@ mod runtime_tests { use crate::types_v2; let analyzer = Analyzer::new(); for entry in registry.entries() { + // Mirrors main.rs's QuerySpec construction post-B3/B4: + // thread grouping_labels into group_by_labels; let the + // parser drive time_window when query_string is present. let spec = QuerySpec { query_string: entry.query_string.clone(), metric_name: entry.metric_name.clone(), label_filters: Default::default(), - group_by_labels: vec![], + group_by_labels: entry.grouping_labels.clone(), aggregations: vec!["quantile".into()], - time_window: "5m".into(), + time_window: if entry.query_string.is_some() { + String::new() + } else { + "5m".into() + }, repeat_every: None, accuracy_sla: entry.accuracy_sla, latency_sla: None, @@ -602,4 +609,114 @@ mod runtime_tests { "routing table should have 5 entries (5 sketches; raw declines), got: {map:?}" ); } + + // ── B3 regression: WorkloadEntry.grouping_labels populates emit ─────── + // + // Pre-B3 the WorkloadEntry YAML had no way to declare grouping + // labels — the analyzer pulled them only from PromQL `by (...)` + // clauses. Bare `quantile_over_time(0.99, metric[30s])` carries no + // `by`, so `QueryWorkload.group_by_labels` ended up empty, so + // `collect_metric_to_grouping_labels` returned `{metric: vec![]}`, + // so the 5-sketch routing emitter wrote + // `keep_keys(datapoint.attributes, [])` — stripping ALL attrs + // instead of keeping `["zone"]`. Sid catalog ended up with one sid + // per metric instead of one per (metric × zone). + // + // Post-B3 a declarative `grouping_labels: [zone]` on WorkloadEntry + // is threaded through the pre-pop QuerySpec → analyzer → + // QueryWorkload.group_by_labels → collect_metric_to_grouping_labels + // → the emitter's keep_keys list. Without this round-trip the + // smoke test's sid catalog stays empty-per-zone. + #[test] + fn workload_entry_grouping_labels_round_trip_through_emit_to_keep_keys() { + 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 + grouping_labels: ["zone"] +"#; + let entries: Vec = + serde_yaml::from_str(yaml).expect("parse workload yaml"); + assert_eq!(entries.len(), 1); + assert_eq!( + entries[0].grouping_labels, + vec!["zone".to_string()], + "WorkloadEntry must surface grouping_labels from YAML" + ); + + let registry = crate::workload::WorkloadRegistry::from_entries(entries); + let store = WorkloadStore::new(); + populate_store_from_registry(®istry, &store); + + // The analyzer must have threaded grouping_labels into + // QueryWorkload.group_by_labels. + let map = collect_metric_to_grouping_labels(®istry, &store); + assert_eq!( + map.get("http_requests_total_latency_ms"), + Some(&vec!["zone".to_string()]), + "collect_metric_to_grouping_labels must surface entry.grouping_labels — \ + without this the agent strips ALL attrs and the sid catalog ends up \ + with one sid per metric instead of one per (metric, zone)\nmap: {map:?}" + ); + } + + /// Belt-and-braces companion: the emit-side keep_keys statement + /// must contain the per-entry grouping_labels VERBATIM. Catches a + /// regression where the pre-pop loop populates the workload store + /// but the round-trip through the emitter drops the labels. + #[test] + fn workload_entry_grouping_labels_surface_in_emit_keep_keys_list() { + use crate::physical::colored_dag::emitter::{EdgeStageConfig, ExportTarget}; + use crate::physical::colored_dag::stage_id::StageId; + use crate::sketch_algebra::params::SketchKind; + + 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 + grouping_labels: ["zone"] +"#; + 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 mut edge_cfg = EdgeStageConfig { + source_metric: Some("http_requests_total_latency_ms".to_string()), + label_filters: Vec::new(), + window_secs: Some(30), + sketch_processors: Vec::new(), + exporter_target: ExportTarget::Stage(StageId::Gateway), + prometheus_archive_metrics: Vec::new(), + archive_tier_metrics: Vec::new(), + warm_passthrough_metrics: Vec::new(), + metric_to_family: std::collections::HashMap::from([( + "http_requests_total_latency_ms".to_string(), + SketchKind::DDSketch, + )]), + metric_to_grouping_labels: std::collections::HashMap::new(), + }; + edge_cfg.metric_to_grouping_labels = collect_metric_to_grouping_labels(®istry, &store); + + let yaml_out = crate::emit::emit_edge_yaml(&edge_cfg, "ws://c/", "test-agent") + .expect("emit ok"); + assert!( + yaml_out.contains( + "keep_keys(datapoint.attributes, [\"zone\"]) where metric.name == \"http_requests_total_latency_ms\"" + ), + "keep_keys must list `zone` (NOT empty) for the YAML-declared grouping_labels\n{yaml_out}", + ); + // Belt-and-braces: the bug surface is specifically + // `keep_keys(..., [])`. Make sure we don't accidentally emit + // the empty-list form for this metric. + assert!( + !yaml_out.contains( + "keep_keys(datapoint.attributes, []) where metric.name == \"http_requests_total_latency_ms\"" + ), + "empty keep_keys would strip all attrs and break per-zone sid splitting\n{yaml_out}", + ); + } } diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index c4e2f5ba..866c3dde 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -108,6 +108,45 @@ struct Pipeline { exporters: Vec, } +// ── Window-size clamping (MVP blocker B4) ───────────────────────────────────── +// +// The controller derives `window_secs` from the workload's matrix-selector +// range (`metric[30s]` → 30s). Two bounds keep the emitted value sane: +// +// * Lower bound 5s — below this the sketch processor mints a new +// window before it has enough samples for the family's quality +// guarantees, and the per-flush cardinality on the sid catalog +// explodes (one (sid, window) row per few seconds). +// * Upper bound 60s — above this the user's query range no longer +// contains a closed sketch window, and replay queries return NoData +// while the warm tier still owns the metric. 60s is also the +// historical default the legacy single-pipeline emitter shipped with, +// so clamping here preserves backwards-compat for plans without an +// explicit range. +// +// `None` means "no `Window` node in the typed L5 — agent runs in batch +// mode, no window_duration in the YAML"; we pass that straight through. +// +// Centralised here so [`build_edge_processor_block`] (sketch processor +// `window_duration`) and the [`BackendAggregation`] consumer +// ([`emit_backend_streaming_config_json`]) clamp to the same bounds. The +// downstream backend's reducer keys windows by the emitted value, so +// the two MUST agree or replay-vs-warm answers go out of sync. + +/// Lower bound for [`clamp_window_secs`]. +pub const MIN_WINDOW_SECS: u64 = 5; + +/// Upper bound for [`clamp_window_secs`]. Matches the legacy default +/// the pre-B4 emitter shipped with. +pub const MAX_WINDOW_SECS: u64 = 60; + +/// Clamp a derived window size to `[MIN_WINDOW_SECS, MAX_WINDOW_SECS]`. +/// `None` is preserved as `None` so callers can keep the +/// "no-window / batch-mode" branch distinguishable from a clamped value. +pub fn clamp_window_secs(w: Option) -> Option { + w.map(|s| s.clamp(MIN_WINDOW_SECS, MAX_WINDOW_SECS)) +} + // ── Public API ──────────────────────────────────────────────────────────────── /// Build the OTel-collector YAML for an edge agent from the typed L5 @@ -164,9 +203,14 @@ pub fn emit_edge_yaml( let mut processors: BTreeMap = BTreeMap::new(); let mut sketch_pipeline_processors: Vec = Vec::new(); for sp in &cfg.sketch_processors { + // MVP blocker B4: clamp `window_secs` to [5, 60] so the agent's + // sketch processor's `window_duration` always sits inside the + // user's query range. Without this, `metric[5m]` lands a + // 300s window which is larger than any sensible replay range + // and produces NoData under `quantile_over_time`. let block = build_edge_processor_block( sp, - cfg.window_secs, + clamp_window_secs(cfg.window_secs), &cfg.label_filters, cfg.source_metric.as_deref(), ); @@ -952,10 +996,15 @@ fn emit_edge_yaml_5sketch_routing( } }) .min(); + // MVP blocker B4: clamp `window_secs` to [5, 60] on the + // 5-sketch routing path too. Without this every per-family + // processor in the routed YAML inherits the unclamped 300s + // window from `[5m]` queries. + let clamped_window = clamp_window_secs(cfg.window_secs); let block = if let Some(sp) = family_to_proc.get(&kind) { - build_edge_processor_block(sp, cfg.window_secs, &cfg.label_filters, metric_name_hint) + build_edge_processor_block(sp, clamped_window, &cfg.label_filters, metric_name_hint) } else { - build_default_edge_processor_block(&kind, cfg.window_secs, metric_name_hint) + build_default_edge_processor_block(&kind, clamped_window, metric_name_hint) }; processors.insert(processor_name.to_string(), block); } @@ -1590,6 +1639,17 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue { AggregationInput::SketchEnvelope => "sketch_envelope", AggregationInput::Raw => "raw", }; + // MVP blocker B4: clamp `windowSize` so the backend's reducer keys + // windows by the SAME size the agent's sketch processor uses. The + // backend's `streaming-config.window_size` must match the agent's + // `window_duration` exactly — drift here de-syncs the warm tier + // and replay queries return NoData (the backend's pre-compute + // engine looks for closed windows at the streaming-config size). + // `agg.window_secs` is u64 (not Option) here; passing through + // `clamp_window_secs(Some(_))` and unwrapping keeps the contract + // explicit. + let window_size = clamp_window_secs(Some(agg.window_secs)) + .expect("clamp_window_secs preserves Some"); json!({ "aggregationType": sketch_kind_to_backend_type(&agg.sketch_kind, &agg.sketch_params), "aggregationSubType": "", @@ -1600,7 +1660,7 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue { "aggregated": Vec::::new(), }, "parameters": parameters, - "windowSize": agg.window_secs, + "windowSize": window_size, "windowType": "tumbling", "spatialFilter": agg.spatial_filter, "aggregationInput": aggregation_input, @@ -3761,6 +3821,132 @@ mod tests { /// write destination). We assert ABSENCE line-by-line so the /// `tsdb_bucket:` line (which contains the substring `bucket:`) /// doesn't falsely trigger the negative match. + // ── MVP blocker B4: window-size clamp tests ─────────────────────────── + // + // The controller derives `window_secs` from the workload's matrix- + // selector range. Tests below pin the clamp contract: + // * `[30s]` (sensible inner range) → passes through unchanged + // * `[5m]` (300s) → clamps DOWN to MAX_WINDOW_SECS (60) + // * no `[range]` (analyzer's 5m fallback at the spec level) → + // also clamps DOWN to 60 + // * `[1s]` (below floor) → clamps UP to MIN_WINDOW_SECS (5) + // * `None` (batch mode, no Window node) → stays None + // + // Both consumers must agree (sketch processor's window_duration in + // the agent YAML AND BackendAggregation's windowSize in the + // streaming-config JSON), otherwise the backend's reducer keys + // windows by a size the agent never closes. + + #[test] + fn b4_clamp_window_secs_in_range_passes_through() { + assert_eq!(clamp_window_secs(Some(30)), Some(30)); + assert_eq!(clamp_window_secs(Some(MIN_WINDOW_SECS)), Some(MIN_WINDOW_SECS)); + assert_eq!(clamp_window_secs(Some(MAX_WINDOW_SECS)), Some(MAX_WINDOW_SECS)); + } + + #[test] + fn b4_clamp_window_secs_above_max_clamps_down() { + assert_eq!(clamp_window_secs(Some(300)), Some(MAX_WINDOW_SECS)); + assert_eq!(clamp_window_secs(Some(3600)), Some(MAX_WINDOW_SECS)); + } + + #[test] + fn b4_clamp_window_secs_below_min_clamps_up() { + assert_eq!(clamp_window_secs(Some(0)), Some(MIN_WINDOW_SECS)); + assert_eq!(clamp_window_secs(Some(1)), Some(MIN_WINDOW_SECS)); + assert_eq!(clamp_window_secs(Some(4)), Some(MIN_WINDOW_SECS)); + } + + #[test] + fn b4_clamp_window_secs_none_passes_through() { + assert_eq!(clamp_window_secs(None), None); + } + + /// Pre-B4: a `[5m]` workload landed `window_duration: 300s` in the + /// emitted YAML. Post-B4 the clamp brings it down to 60s so the + /// sketch processor's window sits inside any sensible replay range. + #[test] + fn b4_edge_yaml_clamps_oversize_window_duration() { + let mut cfg = ddsketch_edge_cfg(); + cfg.window_secs = Some(300); // [5m] in the workload + let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); + assert!( + yaml.contains("window_duration: 60s"), + "5m window must clamp to 60s in the sketch processor block\n{yaml}" + ); + assert!( + !yaml.contains("window_duration: 300s"), + "unclamped 300s window must not be emitted\n{yaml}" + ); + } + + #[test] + fn b4_edge_yaml_preserves_inrange_window_duration() { + let mut cfg = ddsketch_edge_cfg(); + cfg.window_secs = Some(30); // [30s] — canonical MVP query range + let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); + assert!( + yaml.contains("window_duration: 30s"), + "30s window is inside [5, 60] and must pass through\n{yaml}" + ); + } + + /// 5-sketch routing path applies the same clamp — every per-family + /// processor inherits the clamped window. + #[test] + fn b4_5sketch_routing_clamps_window_duration_across_all_families() { + let mut cfg = five_sketch_edge_cfg(); + cfg.window_secs = Some(300); + let yaml = emit_edge_yaml(&cfg, "ws://c/", "test-agent").expect("emit ok"); + assert!( + !yaml.contains("window_duration: 300s"), + "5-sketch routing must not leak unclamped 300s windows\n{yaml}", + ); + // The 5-sketch path emits the same processor key 5x (one per + // family); at least one must show the clamped value. + let clamped_count = yaml.matches("window_duration: 60s").count(); + assert!( + clamped_count >= 1, + "5-sketch routing must emit clamped window_duration\n{yaml}", + ); + } + + /// Streaming-config JSON `windowSize` clamps too, so the backend + /// reducer keys windows by the SAME size the agent's sketch + /// processor closes. Drift here de-syncs warm tier replay answers. + #[test] + fn b4_streaming_config_clamps_window_size() { + use crate::physical::colored_dag::emitter::{ + AggregationInput, BackendAggregation, BackendReadout, BackendStageConfig, + }; + use crate::sketch_algebra::params::DDSketchParams; + use crate::sketch_algebra::physical_expr::EstimateOp; + + let cfg = BackendStageConfig { + aggregations: vec![BackendAggregation { + aggregation_id: "agg0".to_string(), + metric_name: "http_requests_total_latency_ms".to_string(), + sketch_kind: SketchKind::DDSketch, + sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), + window_secs: 300, // pre-clamp 5m + spatial_filter: String::new(), + grouping: vec!["zone".to_string()], + aggregation_input: AggregationInput::SketchEnvelope, + }], + readouts: vec![BackendReadout { + aggregation_id: "agg0".to_string(), + op: EstimateOp::Quantile { q: 0.99 }, + }], + }; + let v = emit_backend_streaming_config_json(&cfg).expect("emit ok"); + let aggs = v.get("aggregations").and_then(|a| a.as_array()).expect("aggregations"); + assert_eq!( + aggs[0]["windowSize"].as_u64(), + Some(MAX_WINDOW_SECS), + "windowSize must clamp 300 → 60 so backend reducer matches the agent's emitted window\n{v}" + ); + } + #[test] fn b1_gorillas3_emit_drops_bucket_field_keeps_tsdb_bucket() { let yaml = build_gorillas3_yaml(60); diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 77db6373..f3ccea66 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -269,13 +269,42 @@ async fn main() { { let analyzer = Analyzer::new(); for entry in workload_registry.entries() { + // MVP blocker B4 — let the analyzer parse `time_window` from + // `query_string` (matrix-selector `[range]`) instead of + // forcing a hardcoded "5m" default that overrides whatever + // the user wrote. The analyzer falls back to its own 5m + // default when the PromQL has no matrix selector (e.g. + // `count(unique_users_per_min)`), so this is strictly an + // improvement for queries that DO carry an explicit range. + // Empty string here means "no override; trust the parsed + // value or the analyzer's fallback". + // + // MVP blocker B3 — thread the WorkloadEntry's declarative + // `grouping_labels` into `QuerySpec.group_by_labels`. The + // analyzer merges these with any `by (...)` keys the + // PromQL parser surfaces, populating `QueryWorkload. + // group_by_labels`, which `collect_metric_to_grouping_labels` + // then drops into `EdgeStageConfig.metric_to_grouping_labels` + // so the agent's `keep_keys(datapoint.attributes, [...])` + // OTTL processor strips wire attrs down to this list + // BEFORE sketching. let spec = pipeline::QuerySpec { query_string: entry.query_string.clone(), metric_name: entry.metric_name.clone(), label_filters: Default::default(), - group_by_labels: vec![], + group_by_labels: entry.grouping_labels.clone(), aggregations: vec!["quantile".into()], - time_window: "5m".into(), + // Empty when the entry HAS a `query_string` (the parser + // surfaces the matrix-selector range or its own 5m + // fallback). For entries without a query_string we + // can't trust the parser, so fall back to the + // historical 5m default so the analyzer doesn't error + // out at Step 4. + time_window: if entry.query_string.is_some() { + String::new() + } else { + "5m".into() + }, repeat_every: None, accuracy_sla: entry.accuracy_sla, latency_sla: None, diff --git a/control_plane/src/workload.rs b/control_plane/src/workload.rs index 7e638dca..3492a2bb 100644 --- a/control_plane/src/workload.rs +++ b/control_plane/src/workload.rs @@ -41,6 +41,29 @@ pub struct WorkloadEntry { /// integration). Not yet read by the planner. #[serde(default)] pub target_path: Option, + /// MVP blocker B3 — declarative grouping labels for the wire-attr + /// allowlist the agent applies before sketching. Mirrors the + /// streaming-config's `grouping_labels` contract: the agent's + /// `transform/keep_for_` OTTL processor calls + /// `keep_keys(datapoint.attributes, [...])` on this list, stripping + /// every other attr BEFORE the sketch processor mints sids. + /// + /// Why this is a separate field (not parsed from `query_string`): + /// the canonical MVP workload `quantile_over_time(0.99, + /// http_requests_total_latency_ms[30s])` carries no `by (...)` + /// clause, so the PromQL parser surfaces an EMPTY group_by_labels. + /// Without a declarative field the analyzer ends up with an empty + /// `QueryWorkload.group_by_labels` → an empty `keep_keys` list → + /// the agent strips ALL attrs and mints a single sid per metric + /// (instead of one per `(metric, zone)`), defeating the streaming- + /// config contract. + /// + /// Threaded into `QueryWorkload::group_by_labels` by the registry + /// pre-pop loop in `main`, so it merges with any `by (...)` keys + /// the PromQL parser surfaces. Empty / missing ⇒ same behaviour as + /// pre-B3 (no allowlist injected). + #[serde(default)] + pub grouping_labels: Vec, } fn default_accuracy_sla() -> f64 { @@ -187,6 +210,7 @@ mod tests { assign_to_role: "agent".into(), sketch_family_override: None, target_path: None, + grouping_labels: vec![], }, WorkloadEntry { metric_name: "b".into(), @@ -195,6 +219,7 @@ mod tests { assign_to_role: "backend".into(), sketch_family_override: None, target_path: None, + grouping_labels: vec![], }, WorkloadEntry { metric_name: "c".into(), @@ -203,6 +228,7 @@ mod tests { assign_to_role: "agent".into(), sketch_family_override: None, target_path: None, + grouping_labels: vec![], }, ], };