From d314642660784eb9be0486950acee6d90ecbecc0 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 12 May 2026 20:17:18 -0600 Subject: [PATCH] feat(asap_types): make aggregationId optional; backend derives from content MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 5 M2.1 — first half of the controller→backend aggregation_id handoff. The YAML deserializers (`AggregationConfig::from_yaml_data`, `::deserialize_from_json`, `StreamingConfig::from_yaml_data`) now accept entries without an `aggregationId` field. When absent, the backend computes a deterministic u64 via `compute_agg_config_id(metric, agg_type, sub_type, parameters, grouping_labels)` (xxh64 over a canonical encoding — parameters are BTreeMap-sorted so insertion order doesn't matter). Backwards compatible: existing YAML that includes `aggregationId` still parses and the explicit value is honored unchanged. This is the consumer-side prep for M2.2, which will drop the field from `controller::emit::asapquery_backend`'s output. Co-Authored-By: Claude Opus 4.7 (1M context) --- Cargo.lock | 1 + crates/asap_types/Cargo.toml | 1 + crates/asap_types/src/aggregation_config.rs | 185 +++++++++++++++++++- crates/asap_types/src/streaming_config.rs | 52 +++--- 4 files changed, 212 insertions(+), 27 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 57791cc5..21f39a5e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -414,6 +414,7 @@ dependencies = [ "serde_json", "serde_yaml", "tracing", + "xxhash-rust", ] [[package]] diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index 0c74b384..ed316da6 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -11,3 +11,4 @@ serde_json.workspace = true serde_yaml.workspace = true anyhow.workspace = true clap.workspace = true +xxhash-rust = { version = "0.8", features = ["xxh64"] } diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index 52c12d92..0bd7985a 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -1,13 +1,14 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use serde_yaml; -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use crate::enums::{QueryLanguage, WindowType}; use crate::traits::SerializableToSink; use crate::utils::normalize_spatial_filter; use promql_utilities::data_model::KeyByLabelNames; use promql_utilities::query_logics::enums::AggregationType; +use xxhash_rust::xxh64::xxh64; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AggregationConfig { @@ -47,6 +48,61 @@ pub struct AggregationIdInfo { // TODO: need to implement deserialization methods +/// Derive a stable `aggregation_id` from the agg-config content when the +/// controller-emitted YAML omits the `aggregationId` field. Phase 5 M2 +/// follow-up: same (metric, agg_type, sub_type, parameters, +/// grouping_labels) tuple always yields the same id, so the controller +/// no longer needs to mint one. xxh64 keeps the id portable across +/// hosts (vs `std::hash::DefaultHasher`, which is not stable). +/// +/// Parameters are canonicalized via BTreeMap so map iteration order +/// doesn't affect the result. The fingerprint format is private to +/// this function — never persisted, never compared across versions. +/// +/// 0 is reserved as a sentinel in legacy test fixtures; if a real +/// input hashes to 0 (vanishingly unlikely), we perturb to 1. +pub fn compute_agg_config_id( + metric: &str, + aggregation_type: &AggregationType, + aggregation_sub_type: &str, + parameters: &HashMap, + grouping_labels: &KeyByLabelNames, +) -> u64 { + let mut buf: Vec = Vec::new(); + buf.extend_from_slice(metric.as_bytes()); + buf.push(0); + // AggregationType: Serialize impl yields a stable string repr. + buf.extend_from_slice( + serde_json::to_string(aggregation_type) + .unwrap_or_default() + .as_bytes(), + ); + buf.push(0); + buf.extend_from_slice(aggregation_sub_type.as_bytes()); + buf.push(0); + // Canonicalize parameters: sort by key, render each value via + // serde_json so nested structure is encoded deterministically. + let sorted: BTreeMap<&String, &Value> = parameters.iter().collect(); + for (k, v) in sorted { + buf.extend_from_slice(k.as_bytes()); + buf.push(b'='); + buf.extend_from_slice(serde_json::to_string(v).unwrap_or_default().as_bytes()); + buf.push(b';'); + } + buf.push(0); + // grouping_labels.labels is already sorted at construction. + for l in &grouping_labels.labels { + buf.extend_from_slice(l.as_bytes()); + buf.push(b','); + } + let h = xxh64(&buf, 0); + if h == 0 { + 1 + } else { + h + } +} + impl AggregationConfig { #[allow(clippy::too_many_arguments)] pub fn new( @@ -110,9 +166,10 @@ impl AggregationConfig { pub fn deserialize_from_json( data: &Value, ) -> Result> { - let aggregation_id = data["aggregationId"] - .as_u64() - .ok_or("Missing aggregationId")?; + // M2 follow-up — `aggregationId` is now optional. When the + // controller-emitted YAML omits it, we derive a deterministic + // id from the agg-config content. + let explicit_id = data["aggregationId"].as_u64(); let aggregation_type: AggregationType = data["aggregationType"] .as_str() @@ -171,6 +228,16 @@ impl AggregationConfig { .and_then(|v| v.as_str()) .map(|s| s.to_string()); + let aggregation_id = explicit_id.unwrap_or_else(|| { + compute_agg_config_id( + &metric, + &aggregation_type, + &aggregation_sub_type, + ¶meters, + &grouping_labels, + ) + }); + Ok(Self::new( aggregation_id, aggregation_type, @@ -204,9 +271,9 @@ impl AggregationConfig { num_aggregates_to_retain: Option, query_language: QueryLanguage, ) -> Result { - let aggregation_id = aggregation_data["aggregationId"] - .as_u64() - .ok_or_else(|| anyhow::anyhow!("Missing aggregationId"))?; + // M2 follow-up — `aggregationId` is optional. When the + // controller-emitted YAML omits it, derive from content below. + let explicit_id = aggregation_data["aggregationId"].as_u64(); let labels = &aggregation_data["labels"]; let grouping_labels = KeyByLabelNames::new( @@ -292,6 +359,16 @@ impl AggregationConfig { } }; + let aggregation_id = explicit_id.unwrap_or_else(|| { + compute_agg_config_id( + &metric, + &aggregation_type, + &aggregation_sub_type, + ¶meters, + &grouping_labels, + ) + }); + Ok(Self::new( aggregation_id, aggregation_type, @@ -348,3 +425,97 @@ impl SerializableToSink for AggregationConfig { self.original_yaml.as_bytes().to_vec() } } + +#[cfg(test)] +mod tests { + use super::*; + + fn sample_yaml(with_id: bool) -> serde_yaml::Value { + let id_line = if with_id { "aggregationId: 42\n" } else { "" }; + let yaml = format!( + "{id_line}aggregationType: DDSketch\naggregationSubType: ''\nmetric: http_latency_ms\nlabels:\n grouping: [zone]\n rollup: []\n aggregated: []\nparameters:\n relative_accuracy: 0.01\nwindowSize: 30\nwindowType: tumbling\nspatialFilter: ''\n", + id_line = id_line + ); + serde_yaml::from_str(&yaml).expect("yaml parses") + } + + #[test] + fn explicit_aggregation_id_is_honored() { + let cfg = AggregationConfig::from_yaml_data( + &sample_yaml(true), + None, + QueryLanguage::promql, + ) + .expect("parse ok"); + assert_eq!(cfg.aggregation_id, 42); + } + + #[test] + fn missing_aggregation_id_is_derived_deterministically() { + let a = AggregationConfig::from_yaml_data( + &sample_yaml(false), + None, + QueryLanguage::promql, + ) + .expect("parse without id"); + let b = AggregationConfig::from_yaml_data( + &sample_yaml(false), + None, + QueryLanguage::promql, + ) + .expect("parse without id again"); + assert_eq!( + a.aggregation_id, b.aggregation_id, + "same content yields same derived id" + ); + assert_ne!(a.aggregation_id, 0, "derived id is never the 0 sentinel"); + } + + #[test] + fn derived_id_changes_with_metric() { + let mut params = HashMap::new(); + params.insert("relative_accuracy".to_string(), serde_json::json!(0.01)); + let grouping = KeyByLabelNames::new(vec!["zone".to_string()]); + let a = compute_agg_config_id( + "http_latency_ms", + &AggregationType::DDSketch, + "", + ¶ms, + &grouping, + ); + let b = compute_agg_config_id( + "cpu_seconds", + &AggregationType::DDSketch, + "", + ¶ms, + &grouping, + ); + assert_ne!(a, b); + } + + #[test] + fn derived_id_changes_with_parameters() { + let grouping = KeyByLabelNames::new(vec!["zone".to_string()]); + let mut p1 = HashMap::new(); + p1.insert("relative_accuracy".to_string(), serde_json::json!(0.01)); + let mut p2 = HashMap::new(); + p2.insert("relative_accuracy".to_string(), serde_json::json!(0.005)); + let a = compute_agg_config_id("m", &AggregationType::DDSketch, "", &p1, &grouping); + let b = compute_agg_config_id("m", &AggregationType::DDSketch, "", &p2, &grouping); + assert_ne!(a, b); + } + + #[test] + fn derived_id_independent_of_parameters_insertion_order() { + let grouping = KeyByLabelNames::new(vec!["zone".to_string()]); + let mut p_ab = HashMap::new(); + p_ab.insert("alpha".to_string(), serde_json::json!(1)); + p_ab.insert("beta".to_string(), serde_json::json!(2)); + let mut p_ba = HashMap::new(); + p_ba.insert("beta".to_string(), serde_json::json!(2)); + p_ba.insert("alpha".to_string(), serde_json::json!(1)); + let a = compute_agg_config_id("m", &AggregationType::DDSketch, "", &p_ab, &grouping); + let b = compute_agg_config_id("m", &AggregationType::DDSketch, "", &p_ba, &grouping); + assert_eq!(a, b); + } +} diff --git a/crates/asap_types/src/streaming_config.rs b/crates/asap_types/src/streaming_config.rs index 13b2b25e..9a427b9e 100644 --- a/crates/asap_types/src/streaming_config.rs +++ b/crates/asap_types/src/streaming_config.rs @@ -82,26 +82,20 @@ impl StreamingConfig { if let Some(aggregations) = data.get("aggregations").and_then(|v| v.as_sequence()) { for aggregation_data in aggregations { - if let Some(aggregation_id) = aggregation_data.get("aggregationId") { - let aggregation_id_u64 = aggregation_id.as_u64().ok_or_else(|| { - anyhow::anyhow!( - "aggregationId must be a valid u64, got: {:?}", - aggregation_id - ) - })?; - // Read per-agg retention directly from the YAML - // entry (previously was looked up via - // inference_config's query→agg map). - let num_aggregates_to_retain = aggregation_data - .get("numAggregatesToRetain") - .and_then(|v| v.as_u64()); - let config = AggregationConfig::from_yaml_data( - aggregation_data, - num_aggregates_to_retain, - QueryLanguage::promql, - )?; - aggregation_configs.insert(aggregation_id_u64, config); - } + // Read per-agg retention directly from the YAML entry + // (previously was looked up via inference_config's + // query→agg map). `aggregationId` is no longer + // required — `AggregationConfig::from_yaml_data` + // derives it from content when absent (M2 follow-up). + let num_aggregates_to_retain = aggregation_data + .get("numAggregatesToRetain") + .and_then(|v| v.as_u64()); + let config = AggregationConfig::from_yaml_data( + aggregation_data, + num_aggregates_to_retain, + QueryLanguage::promql, + )?; + aggregation_configs.insert(config.aggregation_id, config); } } @@ -154,4 +148,22 @@ mod tests { let cfg: StreamingConfig = serde_json::from_str(yaml).expect("Phase-5 decode"); assert_eq!(cfg.storage_backend(), StorageBackend::GorillaObjectStore); } + + /// M2 follow-up: a streaming-config YAML that omits `aggregationId` + /// on every aggregation parses correctly — backend derives the id + /// from the agg-config content. This is the path the controller + /// will use once it stops emitting the field. + #[test] + fn from_yaml_data_accepts_entry_without_aggregation_id() { + let yaml = "\ +aggregations:\n\ +- aggregationType: DDSketch\n aggregationSubType: ''\n metric: cpu_seconds\n labels:\n grouping: [host]\n rollup: []\n aggregated: []\n parameters:\n relative_accuracy: 0.01\n windowSize: 30\n windowType: tumbling\n spatialFilter: ''\n"; + let data: Value = serde_yaml::from_str(yaml).expect("yaml ok"); + let cfg = StreamingConfig::from_yaml_data(&data).expect("decode without id"); + assert_eq!(cfg.aggregation_configs.len(), 1); + let (k, v) = cfg.aggregation_configs.iter().next().unwrap(); + assert_ne!(*k, 0, "derived id is not the 0 sentinel"); + assert_eq!(*k, v.aggregation_id, "map key matches the agg's id"); + assert_eq!(v.metric, "cpu_seconds"); + } }